[WIP] Add Kafka invoke type for Go functions

Makefile

@@ -206,8 +206,10 @@ check-templates: check-go check-rust check-typescript ## Run template source cod
 check-go: ## Check Go templates' source
 	cd templates/go/scaffolding/instanced-http && go vet ./... &&  $(BIN_GOLANGCI_LINT) run
 	cd templates/go/scaffolding/instanced-cloudevents && go vet && $(BIN_GOLANGCI_LINT) run
+	cd templates/go/scaffolding/instanced-kafka && go vet ./... && $(BIN_GOLANGCI_LINT) run
 	cd templates/go/scaffolding/static-http && go vet ./... && $(BIN_GOLANGCI_LINT) run
 	cd templates/go/scaffolding/static-cloudevents && go vet ./... && $(BIN_GOLANGCI_LINT) run
+	cd templates/go/scaffolding/static-kafka && go vet ./... && $(BIN_GOLANGCI_LINT) run
 
 .PHONY: check-rust
 check-rust: ## Check Rust templates' source
@@ -226,6 +228,7 @@ test-templates: test-go test-node test-python test-quarkus test-springboot test-
 test-go: ## Test Go templates
 	cd templates/go/cloudevents && go mod tidy && go test
 	cd templates/go/http && go mod tidy && go test
+	cd templates/go/kafka && go mod tidy && go test
 
 .PHONY: test-node
 test-node: ## Test Node templates
@@ -271,6 +274,8 @@ update-runtime-go:
 	cd templates/go/scaffolding/static-http && go get -u knative.dev/func-go/http
 	cd templates/go/scaffolding/instanced-cloudevents && go get -u knative.dev/func-go/cloudevents
 	cd templates/go/scaffolding/static-cloudevents && go get -u knative.dev/func-go/cloudevents
+	cd templates/go/scaffolding/instanced-kafka && go get -u knative.dev/func-go/kafka
+	cd templates/go/scaffolding/static-kafka && go get -u knative.dev/func-go/kafka
 
 
 .PHONY: certs

cmd/create_test.go

@@ -2,8 +2,12 @@ package cmd
 
 import (
 	"errors"
+	"os"
+	"path/filepath"
 	"testing"
 
+	"gopkg.in/yaml.v2"
+
 	. "knative.dev/func/pkg/testing"
 	"knative.dev/func/pkg/utils"
 )
@@ -82,6 +86,33 @@ func TestCreate_ValidatesName(t *testing.T) {
 	}
 }
 
+// TestCreate_KafkaTemplate ensures that creating a function with the kafka
+// template succeeds and sets invoke to "kafka" in func.yaml.
+func TestCreate_KafkaTemplate(t *testing.T) {
+	_ = FromTempDirectory(t)
+
+	cmd := NewCreateCmd(NewClient)
+	cmd.SetArgs([]string{"--language", "go", "--template", "kafka", "myfunc"})
+
+	if err := cmd.Execute(); err != nil {
+		t.Fatal(err)
+	}
+
+	data, err := os.ReadFile(filepath.Join("myfunc", "func.yaml"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	var funcYaml struct {
+		Invoke string `yaml:"invoke"`
+	}
+	if err := yaml.Unmarshal(data, &funcYaml); err != nil {
+		t.Fatal(err)
+	}
+	if funcYaml.Invoke != "kafka" {
+		t.Fatalf("expected invoke to be 'kafka', got '%s'", funcYaml.Invoke)
+	}
+}
+
 // TestCreate_ConfigOptional ensures that the system can be used without
 // any additional configuration being required.
 func TestCreate_ConfigOptional(t *testing.T) {

cmd/templates_test.go

@@ -24,6 +24,7 @@ func TestTemplates_Default(t *testing.T) {
 	expected := `LANGUAGE     TEMPLATE
 go           cloudevents
 go           http
+go           kafka
 node         cloudevents
 node         http
 python       cloudevents
@@ -57,7 +58,8 @@ func TestTemplates_JSON(t *testing.T) {
 	expected := `{
   "go": [
     "cloudevents",
-    "http"
+    "http",
+    "kafka"
   ],
   "node": [
     "cloudevents",
@@ -105,7 +107,8 @@ func TestTemplates_ByLanguage(t *testing.T) {
 	}
 
 	expected := `cloudevents
-http`
+http
+kafka`
 
 	output := buf()
 	if output != expected {
@@ -121,7 +124,8 @@ http`
 
 	expected = `[
   "cloudevents",
-  "http"
+  "http",
+  "kafka"
 ]`
 
 	output = buf()

docs/reference/func_create.md

@@ -31,6 +31,7 @@ DESCRIPTION
 	  --------     --------
 	  go           cloudevents
 	  go           http
+	  go           kafka
 	  node         cloudevents
 	  node         http
 	  python       cloudevents

pkg/functions/function.go

@@ -107,7 +107,7 @@ type Function struct {
 
 	// Invoke defines hints for use when invoking this function.
 	// See Client.Invoke for usage.
-	Invoke string `yaml:"invoke,omitempty" jsonschema:"enum=http,enum=cloudevent"`
+	Invoke string `yaml:"invoke,omitempty" jsonschema:"enum=http,enum=cloudevent,enum=kafka"`
 
 	// Build defines the build properties for a function
 	Build BuildSpec `yaml:"build,omitempty"`

pkg/functions/templates_test.go

@@ -34,6 +34,7 @@ func TestTemplates_List(t *testing.T) {
 	expected := []string{
 		"cloudevents",
 		"http",
+		"kafka",
 		"customTemplateRepo/customTemplate",
 	}
 

pkg/scaffolding/detectors_test.go

@@ -62,6 +62,24 @@ func New() { }
 			Inv:  "cloudevent",
 			Src: `
 package f
+func Handle() { }
+	`},
+		{
+			Name: "Instanced Kafka",
+			Sig:  InstancedKafka,
+			Err:  nil,
+			Inv:  "kafka",
+			Src: `
+package f
+func New() { }
+	`},
+		{
+			Name: "Static Kafka",
+			Sig:  StaticKafka,
+			Err:  nil,
+			Inv:  "kafka",
+			Src: `
+package f
 func Handle() { }
 	`},
 		{

pkg/scaffolding/signatures.go

@@ -6,17 +6,21 @@ const (
 	UnknownSignature Signature = iota
 	InstancedHTTP
 	InstancedCloudevents
+	InstancedKafka
 	StaticHTTP
 	StaticCloudevents
+	StaticKafka
 )
 
 func (s Signature) String() string {
 	return []string{
 		"unknown",
 		"instanced-http",
 		"instanced-cloudevents",
+		"instanced-kafka",
 		"static-http",
 		"static-cloudevents",
+		"static-kafka",
 	}[s]
 }
 
@@ -28,10 +32,12 @@ func (s Signature) String() string {
 var signatureMap = map[bool]map[string]Signature{
 	true: { // Instanced
 		"http":       InstancedHTTP,
-		"cloudevent": InstancedCloudevents},
-	false: { // !Instanced
+		"cloudevent": InstancedCloudevents,
+		"kafka":      InstancedKafka},
+	false: {
 		"http":       StaticHTTP,
-		"cloudevent": StaticCloudevents},
+		"cloudevent": StaticCloudevents,
+		"kafka":      StaticKafka},
 }
 
 // toSignature converts an instanced boolean and invocation hint into

pkg/scaffolding/signatures_test.go

@@ -26,6 +26,8 @@ func TestSignatures(t *testing.T) {
 		{false, "", StaticHTTP, "static-http"},
 		{false, "http", StaticHTTP, "static-http"},
 		{false, "cloudevent", StaticCloudevents, "static-cloudevents"},
+		{true, "kafka", InstancedKafka, "instanced-kafka"},
+		{false, "kafka", StaticKafka, "static-kafka"},
 		{true, "invalid", UnknownSignature, "unknown"},
 		{false, "invalid", UnknownSignature, "unknown"},
 	}

schema/func_yaml-schema.json

@@ -210,7 +210,8 @@
 				"invoke": {
 					"enum": [
 						"http",
-						"cloudevent"
+						"cloudevent",
+						"kafka"
 					],
 					"type": "string",
 					"description": "Invoke defines hints for use when invoking this function.\nSee Client.Invoke for usage."

templates/go/.instanced-kafka/README.md

@@ -0,0 +1,32 @@
+# Kafka Function Instance
+
+Welcome to your new Go Function! The boilerplate function code can be found in
+[`function.go`](function.go). This Function consumes messages from Kafka topics.
+
+## How it works
+
+Your `Handle` method is called once for each Kafka message. Return `nil` to
+indicate successful processing — the message offset will be committed
+automatically. Return an error to skip the message (the error is logged and
+the consumer moves on to the next message).
+
+## Delivery guarantees
+
+Messages are delivered **at-least-once** and processed **in order per
+partition**. If the consumer crashes after processing a message but before the
+offset is committed, the message will be redelivered. There is no built-in
+deduplication — if your function cannot safely process the same message twice,
+you should implement idempotency in your `Handle` method (for example by
+tracking previously seen message keys or offsets).
+
+## Development
+
+Develop new features by adding a test to [`function_test.go`](function_test.go) for
+each feature, and confirm it works with `go test`.
+
+Once your function is passing tests, deploy it using `func deploy`.  The
+`func` CLI also offers several other testing and development commands; see
+`func --help` for more.
+
+For more, see [the complete documentation]('https://github.com/knative/func/tree/main/docs')
+

templates/go/.instanced-kafka/function.go

@@ -0,0 +1,85 @@
+// package function is an example of a Kafka Function implementation.
+//
+// This package name can be changed when using the "host" builder
+// (as can the module name in go.mod)
+package function
+
+import (
+	"context"
+	"fmt"
+
+	"knative.dev/func-go/kafka"
+)
+
+// MyFunction is the function provided by this library.
+// This structure name can be changed.
+type MyFunction struct{}
+
+// New constructs an instance of your function.  It is called each time a new
+// instance of the function service is created.  This function must be named
+// "New", accept no arguments, and return a structure which exports at least
+// a Handle method (and optionally any of the additional methods described
+// in the comments below).
+func New() *MyFunction {
+	return &MyFunction{}
+}
+
+// Handle a Kafka message.
+//
+// Returning nil signals successful processing and the message offset is
+// committed.  Returning an error logs the error and the message is skipped
+// (not retried).
+func (f *MyFunction) Handle(ctx context.Context, msg kafka.Message) error {
+	fmt.Printf("Received message: topic=%s partition=%d offset=%d key=%s value=%s\n",
+		msg.Topic, msg.Partition, msg.Offset, string(msg.Key), string(msg.Value))
+	return nil
+}
+
+// Start is called whenever a function instance is started.
+//
+// Provided to this start method are all arguments and environment variables
+// which apply to this function.  For better function portability, testability
+// and robustness, it is encouraged to use this method for accessing function
+// configuration rather than looking for environment variables or flags.
+// func (f *MyFunction) Start(ctx context.Context, args map[string]string) error {
+//   fmt.Println("Function Started")
+//   return nil
+// }
+
+// Stop is called whenever a function is stopped.
+//
+// This may happen for reasons such as being rescheduled onto a different node,
+// being updated with a newer version, or if the number of function instances
+// is being scaled down due to low load.  This is a good place to cleanup and
+// realease any resources which expect to be manually released.
+//
+// func (f *Function) Stop(ctx context.Context) error { return nil }
+
+// Alive is an optional method which allows you to more deeply indicate that
+// your function is alive.  The default liveness implementation returns true
+// if the function process is not deadlocked and able to respond.  A custom
+// implementation of this method may be useful when a function should not be
+// considered alive if any dependent services are alive, or other more
+// complex logic.
+//
+// func (f *Function) Alive(ctx context.Context) (bool, error) {
+//   return true, nil
+// }
+
+// Ready is an optional method which, when implemented, will ensure that
+// requests are not made to the Function's request handler until this method
+// reports true.
+//
+// func (f *Function) Ready(ctx context.Context) (bool, error) {
+//   return true, nil
+// }
+
+// Handle is an optional method which can be used to implement simple functions
+// with little or no state, and minimal testing requirements.  By implementing
+// this package static function, one can forego the constructor and struct
+// outlined above.  Note that if this method is defined, the system will ignore
+// the instanced function constructor if it is defined.
+//
+// func Handle(ctx context.Context, msg kafka.Message) error {
+//   /* Your Static Handler Code Here */
+// }

templates/go/.instanced-kafka/function_test.go

@@ -0,0 +1,23 @@
+package function
+
+import (
+	"context"
+	"testing"
+
+	"knative.dev/func-go/kafka"
+)
+
+// TestHandle ensures that the constructor returns an object which handles
+// a Kafka message without error.
+func TestHandle(t *testing.T) {
+	msg := kafka.Message{
+		Key:   []byte("test-key"),
+		Value: []byte("test-value"),
+		Topic: "test-topic",
+	}
+
+	err := New().Handle(context.Background(), msg)
+	if err != nil {
+		t.Fatal(err)
+	}
+}

templates/go/.instanced-kafka/go.mod

@@ -0,0 +1,5 @@
+module function
+
+go 1.21
+
+require knative.dev/func-go v0.21.3

templates/go/.instanced-kafka/go.sum

@@ -0,0 +1 @@
+knative.dev/func-go v0.21.3/go.mod h1:YAUlPi4bY5OQb7n9424zm9GtRigOQ1/IBtqkLqC29Dw=

templates/go/.instanced-kafka/manifest.yaml

@@ -0,0 +1,5 @@
+# optional.  Invocation defines hints for how Functions created using this
+# template can be invoked.  These settings can be updated on the resultant
+# Function as development progresses to ensure 'invoke' can always trigger the
+# execution of a running Function instance for testing and development.
+invoke: "kafka"