fix: publish to old pubsub impl and kafka

# Conflicts:
#	backend/runner/runner.go
#	go-runtime/internal/impl.go

backend/controller/pubsub/integration_test.go

@@ -19,7 +19,7 @@ func TestPubSub(t *testing.T) {
 	calls := 20
 	events := calls * 10
 	in.Run(t,
-		in.WithLanguages("go"),
+		in.WithLanguages("java", "go", "kotlin"),
 		in.CopyModule("publisher"),
 		in.CopyModule("subscriber"),
 		in.Deploy("publisher"),
@@ -45,7 +45,7 @@ func TestPubSub(t *testing.T) {
 
 func TestConsumptionDelay(t *testing.T) {
 	in.Run(t,
-		in.WithLanguages("go"),
+		in.WithLanguages("go", "java"),
 		in.CopyModule("publisher"),
 		in.CopyModule("subscriber"),
 		in.Deploy("publisher"),
@@ -89,7 +89,7 @@ func TestConsumptionDelay(t *testing.T) {
 func TestRetry(t *testing.T) {
 	retriesPerCall := 2
 	in.Run(t,
-		in.WithLanguages("go"),
+		in.WithLanguages("java", "go"),
 		in.CopyModule("publisher"),
 		in.CopyModule("subscriber"),
 		in.Deploy("publisher"),

backend/runner/pubsub/publisher.go

@@ -1,10 +1,16 @@
 package pubsub
 
 import (
+	"context"
 	"fmt"
 
+	"connectrpc.com/connect"
 	"github.com/IBM/sarama"
 
+	schemapb "github.com/TBD54566975/ftl/backend/protos/xyz/block/ftl/schema/v1"
+	ftlv1 "github.com/TBD54566975/ftl/backend/protos/xyz/block/ftl/v1"
+	"github.com/TBD54566975/ftl/backend/protos/xyz/block/ftl/v1/ftlv1connect"
+	"github.com/TBD54566975/ftl/internal/rpc"
 	"github.com/TBD54566975/ftl/internal/schema"
 )
 
@@ -36,7 +42,10 @@ func newPublisher(t *schema.Topic) (*publisher, error) {
 	}, nil
 }
 
-func (p *publisher) publish(data []byte, key string) error {
+func (p *publisher) publish(ctx context.Context, data []byte, key string, caller schema.RefKey) error {
+	if err := p.publishToController(ctx, data, caller); err != nil {
+		return err
+	}
 	_, _, err := p.producer.SendMessage(&sarama.ProducerMessage{
 		Topic: p.topic.Runtime.TopicID,
 		Value: sarama.ByteEncoder(data),
@@ -47,3 +56,19 @@ func (p *publisher) publish(data []byte, key string) error {
 	}
 	return nil
 }
+
+// publishToController publishes the data to the controller (old pubsub implementation)
+//
+// This is to keep pubsub working while we transition fully to Kafka for pubsub.
+func (p *publisher) publishToController(ctx context.Context, data []byte, caller schema.RefKey) error {
+	client := rpc.ClientFromContext[ftlv1connect.ModuleServiceClient](ctx)
+	_, err := client.PublishEvent(ctx, connect.NewRequest(&ftlv1.PublishEventRequest{
+		Topic:  p.topic.ToProto().(*schemapb.Ref), //nolint: forcetypeassert
+		Caller: caller.Name,
+		Body:   data,
+	}))
+	if err != nil {
+		return fmt.Errorf("failed to publish event to controller: %w", err)
+	}
+	return nil
+}

backend/runner/pubsub/pubsub.go

@@ -42,7 +42,11 @@ func (s *Service) PublishEvent(ctx context.Context, req *connect.Request[pb.Publ
 	if !ok {
 		return nil, fmt.Errorf("topic %s not found", req.Msg.Topic.Name)
 	}
-	err := publisher.publish(req.Msg.Body, req.Msg.Key)
+	caller, err := schema.ParseRef(req.Msg.Caller)
+	if err != nil {
+		return nil, fmt.Errorf("could not parse caller: %w", err)
+	}
+	err = publisher.publish(ctx, req.Msg.Body, req.Msg.Key, caller.ToRefKey())
 	if err != nil {
 		return nil, err
 	}

go-runtime/internal/impl.go

@@ -10,8 +10,6 @@ import (
 	"connectrpc.com/connect"
 	"github.com/puzpuzpuz/xsync/v3"
 
-	ftldeployment "github.com/TBD54566975/ftl/backend/protos/xyz/block/ftl/deployment/v1"
-	deploymentconnect "github.com/TBD54566975/ftl/backend/protos/xyz/block/ftl/deployment/v1/ftlv1connect"
 	pubpb "github.com/TBD54566975/ftl/backend/protos/xyz/block/ftl/publish/v1"
 	pubconnect "github.com/TBD54566975/ftl/backend/protos/xyz/block/ftl/publish/v1/publishpbconnect"
 	schemapb "github.com/TBD54566975/ftl/backend/protos/xyz/block/ftl/schema/v1"
@@ -57,32 +55,6 @@ func (r *RealFTL) PublishEvent(ctx context.Context, topic *schema.Ref, event any
 	if topic.Module != caller.Module {
 		return fmt.Errorf("can not publish to another module's topic: %s", topic)
 	}
-	if err := publishToFTL(ctx, topic, event, caller); err != nil {
-		return err
-	}
-	return publishToModule(ctx, topic, event, key, caller)
-}
-
-func publishToFTL(ctx context.Context, topic *schema.Ref, event any, caller schema.RefKey) error {
-	// TODO: remove this once we have other pubsub moved over to kafka
-	// For now we are publishing to both systems.
-	client := rpc.ClientFromContext[deploymentconnect.DeploymentServiceClient](ctx)
-	body, err := encoding.Marshal(event)
-	if err != nil {
-		return fmt.Errorf("failed to marshal event to controller: %w", err)
-	}
-	_, err = client.PublishEvent(ctx, connect.NewRequest(&ftldeployment.PublishEventRequest{
-		Topic:  topic.ToProto().(*schemapb.Ref), //nolint: forcetypeassert
-		Caller: caller.Name,
-		Body:   body,
-	}))
-	if err != nil {
-		return fmt.Errorf("failed to publish event to controller: %w", err)
-	}
-	return nil
-}
-
-func publishToModule(ctx context.Context, topic *schema.Ref, event any, key string, caller schema.RefKey) error {
 	client := rpc.ClientFromContext[pubconnect.PublishServiceClient](ctx)
 	body, err := encoding.Marshal(event)
 	if err != nil {