[WIP] Add Kafka event sourcing for Go functions

pkg/functions/function.go

@@ -176,6 +176,35 @@ type MountSpec struct {
 	Destination string `yaml:"path"`
 }
 
+// KafkaConfig specifies the Kafka event source configuration.
+// When set, the runtime consumes messages from Kafka and delivers them
+// as CloudEvents to the function's handler.
+type KafkaConfig struct {
+	Brokers       string `yaml:"brokers" jsonschema:"description=Comma-separated list of Kafka broker addresses"`
+	Topic         string `yaml:"topic" jsonschema:"description=Kafka topic to consume from"`
+	ConsumerGroup string `yaml:"consumerGroup" jsonschema:"description=Kafka consumer group ID"`
+}
+
+func validateKafka(kafka *KafkaConfig, invoke string) (errors []string) {
+	if kafka == nil {
+		return
+	}
+	if invoke != "cloudevent" {
+		errors = append(errors, "run.kafka is only supported with invoke: cloudevent")
+		return
+	}
+	if kafka.Brokers == "" {
+		errors = append(errors, "run.kafka.brokers is required when Kafka is configured")
+	}
+	if kafka.Topic == "" {
+		errors = append(errors, "run.kafka.topic is required when Kafka is configured")
+	}
+	if kafka.ConsumerGroup == "" {
+		errors = append(errors, "run.kafka.consumerGroup is required when Kafka is configured")
+	}
+	return
+}
+
 // RunSpec
 type RunSpec struct {
 	// List of volumes to be mounted to the function
@@ -189,6 +218,10 @@ type RunSpec struct {
 	// with containerized docker runner and deployed Knative service integration
 	// in development.
 	StartTimeout time.Duration `yaml:"startTimeout,omitempty"`
+
+	// Kafka configures a Kafka event source. When set, the function consumes
+	// messages from the specified Kafka topic and receives them as CloudEvents.
+	Kafka *KafkaConfig `yaml:"kafka,omitempty"`
 }
 
 // DeploySpec
@@ -361,6 +394,7 @@ func (f Function) Validate() error {
 		validateOptions(f.Deploy.Options),
 		ValidateLabels(f.Deploy.Labels),
 		validateGit(f.Build.Git),
+		validateKafka(f.Run.Kafka, f.Invoke),
 	}
 
 	var b strings.Builder

pkg/functions/function_test.go

@@ -13,6 +13,7 @@ import (
 	"github.com/go-git/go-git/v5/config"
 	"github.com/go-git/go-git/v5/plumbing/object"
 	"github.com/google/go-cmp/cmp"
+	"gopkg.in/yaml.v2"
 
 	fn "knative.dev/func/pkg/functions"
 	"knative.dev/func/pkg/mock"
@@ -542,3 +543,158 @@ func TestFunction_LocalTransient(t *testing.T) {
 		t.Fatal("Remote not supposed to be set")
 	}
 }
+
+func TestKafkaConfig_YAMLRoundTrip(t *testing.T) {
+	original := fn.Function{
+		Name:    "test-kafka",
+		Runtime: "go",
+		Run: fn.RunSpec{
+			Kafka: &fn.KafkaConfig{
+				Brokers:       "broker1:9092,broker2:9092",
+				Topic:         "my-topic",
+				ConsumerGroup: "my-group",
+			},
+		},
+	}
+
+	data, err := yaml.Marshal(original)
+	if err != nil {
+		t.Fatalf("failed to marshal: %v", err)
+	}
+
+	var loaded fn.Function
+	if err := yaml.Unmarshal(data, &loaded); err != nil {
+		t.Fatalf("failed to unmarshal: %v", err)
+	}
+
+	if loaded.Run.Kafka == nil {
+		t.Fatal("expected Run.Kafka to be non-nil after round-trip")
+	}
+	if loaded.Run.Kafka.Brokers != "broker1:9092,broker2:9092" {
+		t.Errorf("Brokers: expected %q, got %q", "broker1:9092,broker2:9092", loaded.Run.Kafka.Brokers)
+	}
+	if loaded.Run.Kafka.Topic != "my-topic" {
+		t.Errorf("Topic: expected %q, got %q", "my-topic", loaded.Run.Kafka.Topic)
+	}
+	if loaded.Run.Kafka.ConsumerGroup != "my-group" {
+		t.Errorf("ConsumerGroup: expected %q, got %q", "my-group", loaded.Run.Kafka.ConsumerGroup)
+	}
+}
+
+func TestValidateKafka(t *testing.T) {
+	tests := []struct {
+		name      string
+		kafka     *fn.KafkaConfig
+		invoke    string
+		wantErrs  int
+		wantSubst string
+	}{
+		{
+			name:     "nil kafka config is valid",
+			kafka:    nil,
+			invoke:   "cloudevent",
+			wantErrs: 0,
+		},
+		{
+			name: "valid complete config",
+			kafka: &fn.KafkaConfig{
+				Brokers:       "broker:9092",
+				Topic:         "my-topic",
+				ConsumerGroup: "my-group",
+			},
+			invoke:   "cloudevent",
+			wantErrs: 0,
+		},
+		{
+			name: "wrong invoke type",
+			kafka: &fn.KafkaConfig{
+				Brokers:       "broker:9092",
+				Topic:         "my-topic",
+				ConsumerGroup: "my-group",
+			},
+			invoke:    "http",
+			wantErrs:  1,
+			wantSubst: "only supported with invoke: cloudevent",
+		},
+		{
+			name: "missing brokers",
+			kafka: &fn.KafkaConfig{
+				Topic:         "my-topic",
+				ConsumerGroup: "my-group",
+			},
+			invoke:    "cloudevent",
+			wantErrs:  1,
+			wantSubst: "brokers is required",
+		},
+		{
+			name: "missing topic",
+			kafka: &fn.KafkaConfig{
+				Brokers:       "broker:9092",
+				ConsumerGroup: "my-group",
+			},
+			invoke:    "cloudevent",
+			wantErrs:  1,
+			wantSubst: "topic is required",
+		},
+		{
+			name: "missing consumer group",
+			kafka: &fn.KafkaConfig{
+				Brokers: "broker:9092",
+				Topic:   "my-topic",
+			},
+			invoke:    "cloudevent",
+			wantErrs:  1,
+			wantSubst: "consumerGroup is required",
+		},
+		{
+			name:      "all fields missing",
+			kafka:     &fn.KafkaConfig{},
+			invoke:    "cloudevent",
+			wantErrs:  3,
+			wantSubst: "required",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			f := fn.Function{
+				Root:    t.TempDir(),
+				Name:    "test",
+				Runtime: "go",
+				Invoke:  tt.invoke,
+				Run:     fn.RunSpec{Kafka: tt.kafka},
+			}
+			err := f.Validate()
+			if tt.wantErrs == 0 {
+				if err != nil {
+					t.Errorf("unexpected error: %s", err)
+				}
+			} else {
+				if err == nil {
+					t.Fatalf("expected errors containing %q, got nil", tt.wantSubst)
+				}
+				errStr := err.Error()
+				count := strings.Count(errStr, tt.wantSubst)
+				if count < tt.wantErrs {
+					t.Errorf("expected %d occurrences of %q, got %d (error: %s)", tt.wantErrs, tt.wantSubst, count, errStr)
+				}
+			}
+		})
+	}
+}
+
+func TestKafkaConfig_YAMLOmitEmpty(t *testing.T) {
+	f := fn.Function{
+		Name:    "test-func",
+		Runtime: "go",
+	}
+
+	data, err := yaml.Marshal(f)
+	if err != nil {
+		t.Fatalf("failed to marshal: %v", err)
+	}
+
+	output := string(data)
+	if strings.Contains(output, "kafka:") {
+		t.Errorf("expected YAML output to omit 'kafka:' key when nil, got:\n%s", output)
+	}
+}

pkg/functions/runner.go

@@ -307,6 +307,15 @@ func buildRunnerEnv(job *Job, extras map[string]string) ([]string, error) {
 		env = append(env, k+"="+v)
 	}
 
+	if k := job.Function.Run.Kafka; k != nil && k.Brokers != "" && k.Topic != "" && k.ConsumerGroup != "" {
+		env = append(env,
+			"FUNC_TRANSPORT=kafka",
+			"KAFKA_BROKERS="+k.Brokers,
+			"KAFKA_TOPIC="+k.Topic,
+			"KAFKA_CONSUMER_GROUP="+k.ConsumerGroup,
+		)
+	}
+
 	return env, nil
 }
 

pkg/functions/runner_test.go

@@ -38,3 +38,86 @@ func TestGetRunFuncErrors(t *testing.T) {
 		})
 	}
 }
+
+func TestBuildRunnerEnv_KafkaConfig(t *testing.T) {
+	job := &Job{
+		Function: Function{
+			Root:    t.TempDir(),
+			Runtime: "go",
+			Run: RunSpec{
+				Kafka: &KafkaConfig{
+					Brokers:       "broker1:9092",
+					Topic:         "my-topic",
+					ConsumerGroup: "my-group",
+				},
+			},
+		},
+	}
+	env, err := buildRunnerEnv(job, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	want := map[string]string{
+		"FUNC_TRANSPORT":       "kafka",
+		"KAFKA_BROKERS":        "broker1:9092",
+		"KAFKA_TOPIC":          "my-topic",
+		"KAFKA_CONSUMER_GROUP": "my-group",
+	}
+	for key, val := range want {
+		found := false
+		for _, e := range env {
+			if e == key+"="+val {
+				found = true
+				break
+			}
+		}
+		if !found {
+			t.Errorf("expected %s=%s in env, not found", key, val)
+		}
+	}
+}
+
+func TestBuildRunnerEnv_NoKafka(t *testing.T) {
+	t.Setenv("FUNC_TRANSPORT", "inherited")
+	job := &Job{
+		Function: Function{
+			Root:    t.TempDir(),
+			Runtime: "go",
+		},
+	}
+	env, err := buildRunnerEnv(job, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, e := range env {
+		if e == "FUNC_TRANSPORT=kafka" {
+			t.Error("FUNC_TRANSPORT should not be set to kafka when Kafka is nil")
+		}
+	}
+}
+
+func TestBuildRunnerEnv_KafkaIncomplete(t *testing.T) {
+	t.Setenv("FUNC_TRANSPORT", "inherited")
+	job := &Job{
+		Function: Function{
+			Root:    t.TempDir(),
+			Runtime: "go",
+			Run: RunSpec{
+				Kafka: &KafkaConfig{
+					Brokers:       "broker1:9092",
+					Topic:         "",
+					ConsumerGroup: "my-group",
+				},
+			},
+		},
+	}
+	env, err := buildRunnerEnv(job, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, e := range env {
+		if e == "FUNC_TRANSPORT=kafka" {
+			t.Error("FUNC_TRANSPORT should not be set to kafka when Kafka topic is empty")
+		}
+	}
+}

pkg/k8s/deployer.go

@@ -395,6 +395,7 @@ func (d *Deployer) generateDeployment(f fn.Function, namespace string, daprInsta
 	if err != nil {
 		return nil, fmt.Errorf("failed to process environment variables: %w", err)
 	}
+	envVars = AppendKafkaEnvs(envVars, f.Run.Kafka)
 
 	volumes, volumeMounts, err := ProcessVolumes(f.Run.Volumes, referencedSecrets, referencedConfigMaps, referencedPVCs)
 	if err != nil {
@@ -698,6 +699,19 @@ func withOpenAddress(ee []fn.Env) []fn.Env {
 	return ee
 }
 
+func AppendKafkaEnvs(envVars []corev1.EnvVar, kafka *fn.KafkaConfig) []corev1.EnvVar {
+	if kafka == nil || kafka.Brokers == "" || kafka.Topic == "" || kafka.ConsumerGroup == "" {
+		return envVars
+	}
+	envVars = append(envVars,
+		corev1.EnvVar{Name: "FUNC_TRANSPORT", Value: "kafka"},
+		corev1.EnvVar{Name: "KAFKA_BROKERS", Value: kafka.Brokers},
+		corev1.EnvVar{Name: "KAFKA_TOPIC", Value: kafka.Topic},
+		corev1.EnvVar{Name: "KAFKA_CONSUMER_GROUP", Value: kafka.ConsumerGroup},
+	)
+	return envVars
+}
+
 func createEnvFromSource(value string, referencedSecrets, referencedConfigMaps *sets.Set[string]) (*corev1.EnvFromSource, error) {
 	slices := strings.Split(strings.Trim(value, "{} "), ":")
 	if len(slices) != 2 {