Use Publisher Everywhere

config/armada/config.yaml

@@ -78,6 +78,7 @@ pulsar:
   eventsPrinterSubscription: "EventsPrinter"
   maxAllowedEventsPerMessage: 1000
   maxAllowedMessageSize: 4194304 # 4MB
+  sendTimeout: 5s
   receiverQueueSize: 100
 postgres:
   connection:

config/eventingester/config.yaml

@@ -7,7 +7,6 @@ redis:
 pulsar:
   URL: pulsar://pulsar:6650
   jobsetEventsTopic: events
-  receiveTimeout: 5s
   backoffTime: 1s
   receiverQueueSize: 100
 subscriptionName: "events-ingester"

config/lookoutingesterv2/config.yaml

@@ -10,7 +10,6 @@ metricsPort: 9002
 pulsar:
   URL: "pulsar://pulsar:6650"
   jobsetEventsTopic: "events"
-  receiveTimeout: 5s
   backoffTime: 1s
   receiverQueueSize: 100
 subscriptionName: "lookout-ingester-v2"

config/scheduler/config.yaml

@@ -39,6 +39,7 @@ pulsar:
   compressionLevel: faster
   maxAllowedEventsPerMessage: 1000
   maxAllowedMessageSize: 4194304 #4Mi
+  sendTimeout: 5s
 armadaApi:
   armadaUrl: "server:50051"
   forceNoTls: true

config/scheduleringester/config.yaml

@@ -11,7 +11,6 @@ metrics:
 pulsar:
   URL: "pulsar://localhost:6650"
   jobsetEventsTopic: "events"
-  receiveTimeout: 5s
   backoffTime: 1s
   receiverQueueSize: 100
 subscriptionName: "scheduler-ingester"

internal/armada/configuration/types.go

@@ -70,8 +70,8 @@ type PulsarConfig struct {
 	MaxAllowedEventsPerMessage int `validate:"gte=0"`
 	// Maximum allowed message size in bytes
 	MaxAllowedMessageSize uint
-	// Timeout when polling pulsar for messages
-	ReceiveTimeout time.Duration
+	// Timeout when sending messages asynchronously
+	SendTimeout time.Duration `validate:"required"`
 	// Backoff from polling when Pulsar returns an error
 	BackoffTime time.Duration
 	// Number of pulsar messages that will be queued by the pulsar consumer.

internal/armada/mocks/generate.go

@@ -1,6 +1,5 @@
 package mocks
 
 // Mock implementations used by tests
-//go:generate mockgen -destination=./mock_repository.go -package=mocks "github.com/armadaproject/armada/internal/armada/queue" QueueRepository
-//go:generate mockgen -destination=./mock_deduplicator.go -package=mocks "github.com/armadaproject/armada/internal/armada/submit" Deduplicator,Publisher
+//go:generate mockgen -destination=./mock_deduplicator.go -package=mocks "github.com/armadaproject/armada/internal/armada/submit" Deduplicator
 //go:generate mockgen -destination=./mock_authorizer.go -package=mocks "github.com/armadaproject/armada/internal/armada/server" ActionAuthorizer

internal/armada/server.go

@@ -128,7 +128,7 @@ func Serve(ctx *armadacontext.Context, config *configuration.ArmadaConfig, healt
 		CompressionLevel: config.Pulsar.CompressionLevel,
 		BatchingMaxSize:  config.Pulsar.MaxAllowedMessageSize,
 		Topic:            config.Pulsar.JobsetEventsTopic,
-	}, config.Pulsar.MaxAllowedEventsPerMessage, config.Pulsar.MaxAllowedMessageSize)
+	}, config.Pulsar.MaxAllowedEventsPerMessage, config.Pulsar.MaxAllowedMessageSize, config.Pulsar.SendTimeout)
 	if err != nil {
 		return errors.Wrapf(err, "error creating pulsar producer")
 	}

internal/armada/submit/submit_test.go

@@ -16,14 +16,15 @@ import (
 	"github.com/armadaproject/armada/internal/armada/submit/testfixtures"
 	"github.com/armadaproject/armada/internal/common/armadacontext"
 	"github.com/armadaproject/armada/internal/common/auth/permission"
+	commonMocks "github.com/armadaproject/armada/internal/common/mocks"
 	"github.com/armadaproject/armada/internal/common/util"
 	"github.com/armadaproject/armada/pkg/api"
 	"github.com/armadaproject/armada/pkg/armadaevents"
 	"github.com/armadaproject/armada/pkg/client/queue"
 )
 
 type mockObjects struct {
-	publisher    *mocks.MockPublisher
+	publisher    *commonMocks.MockPublisher
 	queueRepo    *mocks.MockQueueRepository
 	deduplicator *mocks.MockDeduplicator
 	authorizer   *mocks.MockActionAuthorizer
@@ -32,7 +33,7 @@ type mockObjects struct {
 func createMocks(t *testing.T) *mockObjects {
 	ctrl := gomock.NewController(t)
 	return &mockObjects{
-		publisher:    mocks.NewMockPublisher(ctrl),
+		publisher:    commonMocks.NewMockPublisher(ctrl),
 		queueRepo:    mocks.NewMockQueueRepository(ctrl),
 		deduplicator: mocks.NewMockDeduplicator(ctrl),
 		authorizer:   mocks.NewMockActionAuthorizer(ctrl),

internal/common/ingest/ingestion_pipeline_test.go

@@ -347,8 +347,7 @@ func TestRun_LimitsProcessingBatchSize(t *testing.T) {
 func testPipeline(consumer pulsar.Consumer, converter InstructionConverter[*simpleMessages], sink Sink[*simpleMessages]) *IngestionPipeline[*simpleMessages] {
 	return &IngestionPipeline[*simpleMessages]{
 		pulsarConfig: configuration.PulsarConfig{
-			ReceiveTimeout: 10 * time.Second,
-			BackoffTime:    time.Second,
+			BackoffTime: time.Second,
 		},
 		pulsarSubscriptionName: "subscription",
 		pulsarBatchDuration:    batchDuration,

internal/common/mocks/generate.go

@@ -2,4 +2,5 @@ package mocks
 
 // Mock implementations used by tests
 //go:generate mockgen -destination=./mock_pulsar.go -package=mocks "github.com/apache/pulsar-client-go/pulsar" Client,Producer,Message
+//go:generate mockgen -destination=./mock_publisher.go -package=mocks "github.com/armadaproject/armada/internal/common/pulsarutils" Publisher
 //go:generate mockgen -destination=./mock_executorapi.go -package=mocks "github.com/armadaproject/armada/pkg/executorapi" ExecutorApiClient,ExecutorApi_LeaseJobRunsClient