Add messages.max.bytes proprety

etf1 · Aug 28, 2024 · df8b5a4 · df8b5a4
1 parent a069602
commit df8b5a4
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 0 deletions.
diff --git a/config/config.go b/config/config.go
@@ -71,6 +71,7 @@ type Kafka struct {
 	Topic              string `config:"KAFKA_TOPIC"`
 	ProduceChannelSize int    `config:"KAFKA_PRODUCE_CHANNEL_SIZE"`
 	WithDecorators     bool   `config:"KAFKA_WITH_DECORATORS"`
+	MessageMaxBytes    int    `config:"KAFKA_MESSAGE_MAX_BYTES"`
 }
 
 // NewBase returns a new base configuration
@@ -107,6 +108,7 @@ func NewBase(ctx context.Context, configPrefix string) *Base {
 			Topic:              "kafka-mongo-watcher",
 			ProduceChannelSize: 10000,
 			WithDecorators:     true,
+			MessageMaxBytes:    1024 * 1024,
 		},
 	}
 

diff --git a/internal/service/kafka.go b/internal/service/kafka.go
@@ -12,6 +12,7 @@ func (container *Container) GetKafkaProducer() *kafkaconfluent.Producer {
 		producer, err := kafkaconfluent.NewProducer(&kafkaconfluent.ConfigMap{
 			"bootstrap.servers":       container.Cfg.Kafka.BootstrapServers,
 			"go.produce.channel.size": container.Cfg.Kafka.ProduceChannelSize,
+			"message.max.bytes":       container.Cfg.Kafka.MessageMaxBytes,
 		})
 		if err != nil {
 			panic(err)