kafkasink

Documentation

Overview

Example (Full)

package main

import (
	"embed"
	"os"

	eventingv1alpha1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
	"knative.dev/eventing-kafka-broker/test/rekt/resources/kafkasink"

	testlog "knative.dev/reconciler-test/pkg/logging"
	"knative.dev/reconciler-test/pkg/manifest"
)

//go:embed *.yaml
var yaml embed.FS

func main() {
	ctx := testlog.NewContext()
	images := map[string]string{}
	cfg := map[string]interface{}{
		"name":             "foo",
		"namespace":        "bar",
		"topic":            "my-topic",
		"bootstrapServers": []string{"my-bootstrap-server:8082"},
	}
	kafkasink.WithContentMode(eventingv1alpha1.ModeStructured)(cfg)
	kafkasink.WithReplicationFactor(3)(cfg)
	kafkasink.WithNumPartitions(10)(cfg)
	kafkasink.WithAuthSecretName("abc")(cfg)

	files, err := manifest.ExecuteYAML(ctx, yaml, images, cfg)
	if err != nil {
		panic(err)
	}

	manifest.OutputYAML(os.Stdout, files)
}

Output:

apiVersion: eventing.knative.dev/v1alpha1
kind: KafkaSink
metadata:
  name: foo
  namespace: bar
spec:
  topic: "my-topic"
  bootstrapServers:
    - "my-bootstrap-server:8082"
  contentMode: structured
  numPartitions: 10
  replicationFactor: 3
  auth:
    secret:
      ref:
        name: abc

Example (Zero)

package main

import (
	"embed"
	"os"

	testlog "knative.dev/reconciler-test/pkg/logging"
	"knative.dev/reconciler-test/pkg/manifest"
)

//go:embed *.yaml
var yaml embed.FS

func main() {
	ctx := testlog.NewContext()
	images := map[string]string{}
	cfg := map[string]interface{}{
		"name":             "foo",
		"namespace":        "bar",
		"topic":            "my-topic",
		"bootstrapServers": []string{"my-bootstrap-server:8082"},
	}

	files, err := manifest.ExecuteYAML(ctx, yaml, images, cfg)
	if err != nil {
		panic(err)
	}

	manifest.OutputYAML(os.Stdout, files)
}

Output:

apiVersion: eventing.knative.dev/v1alpha1
kind: KafkaSink
metadata:
  name: foo
  namespace: bar
spec:
  topic: "my-topic"
  bootstrapServers:
    - "my-bootstrap-server:8082"

Index

• func Address(ctx context.Context, name string, timings ...time.Duration) (*duckv1.Addressable, error)
• func AsKReference(name string, namespace string) *duckv1.KReference
• func GVR() schema.GroupVersionResource
• func Install(name, topic string, bootstrapServers []string, opts ...manifest.CfgFn) feature.StepFn
• func IsReady(name string, timing ...time.Duration) feature.StepFn
• func ValidateAddress(name string, validate addressable.ValidateAddressFn, timings ...time.Duration) feature.StepFn
• func WithAuthSecretName(name string) manifest.CfgFn
• func WithContentMode(contentMode string) manifest.CfgFn
• func WithNumPartitions(numPartitions int32) manifest.CfgFn
• func WithReplicationFactor(replicationFactor int16) manifest.CfgFn

Examples

• Package (Full)
• Package (Zero)

Functions

func Address

func Address(ctx context.Context, name string, timings ...time.Duration) (*duckv1.Addressable, error)

Address returns a sink's address.

func AsKReference

func AsKReference(name string, namespace string) *duckv1.KReference

func GVR

func GVR() schema.GroupVersionResource

func Install

func Install(name, topic string, bootstrapServers []string, opts ...manifest.CfgFn) feature.StepFn

Install will create a Trigger resource, augmented with the config fn options.

func IsReady

func IsReady(name string, timing ...time.Duration) feature.StepFn

IsReady tests to see if a Trigger becomes ready within the time given.

func ValidateAddress

func ValidateAddress(name string, validate addressable.ValidateAddressFn, timings ...time.Duration) feature.StepFn

ValidateAddress validates the address retured by Address

func WithAuthSecretName

func WithAuthSecretName(name string) manifest.CfgFn

WithAuthSecretName adds the auth secret name

func WithContentMode

func WithContentMode(contentMode string) manifest.CfgFn

WithContentMode adds the content mode

func WithNumPartitions

func WithNumPartitions(numPartitions int32) manifest.CfgFn

WithNumPartitions adds the num of partitions

func WithReplicationFactor

func WithReplicationFactor(replicationFactor int16) manifest.CfgFn

WithReplicationFactor adds the replication factor