ENH - Set message max bytes

README.md

@@ -177,6 +177,11 @@ Configuration variables with prefix are first loaded and then without prefix. Fo
 
 A big value here can increase the heap memory of the application as all the payload that have to be sent to Kafka will be maintained in channel.
 
+#### KAFKA_MESSAGE_MAX_BYTES
+*Type*: integer
+
+*Description*: The maximum message size in bytes at the producer level (default: 1024*1024)
+
 #### LOG_CLI_VERBOSE
 *Type*: boolean
 

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,
 		},
 	}
 

config/config_test.go

@@ -42,6 +42,7 @@ var cfg = &Base{
 		Topic:              "kafka-mongo-watcher",
 		ProduceChannelSize: 10000,
 		WithDecorators:     true,
+		MessageMaxBytes:    1024 * 1024,
 	},
 }
 

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)