bulker: batch_consumer: do intermediate flushes while producing retry…

… messages to avoid overfilling queue.buffering.max.messages
jitsucom · Feb 6, 2025 · 2860ccf · 2860ccf
1 parent 3b6b506
commit 2860ccf
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 4 deletions.
diff --git a/bulkerapp/app/abstract_batch_consumer.go b/bulkerapp/app/abstract_batch_consumer.go
@@ -93,10 +93,11 @@ func NewAbstractBatchConsumer(repository *Repository, destinationId string, batc
 		return nil, abstract.NewError("Error creating consumer: %v", err)
 	}
 	producerConfig := kafka.ConfigMap(utils.MapPutAll(kafka.ConfigMap{
-		"transactional.id": fmt.Sprintf("%s_failed_%s", topicId, config.InstanceId),
-		"batch.size":       config.ProducerBatchSize,
-		"linger.ms":        config.ProducerLingerMs,
-		"compression.type": config.KafkaTopicCompression,
+		"transactional.id":             fmt.Sprintf("%s_failed_%s", topicId, config.InstanceId),
+		"queue.buffering.max.messages": config.ProducerQueueSize,
+		"batch.size":                   config.ProducerBatchSize,
+		"linger.ms":                    config.ProducerLingerMs,
+		"compression.type":             config.KafkaTopicCompression,
 	}, *kafkaConfig))
 
 	bc := &AbstractBatchConsumer{

diff --git a/bulkerapp/app/batch_consumer.go b/bulkerapp/app/batch_consumer.go
@@ -269,6 +269,11 @@ func (bc *BatchConsumerImpl) processFailed(firstPosition *kafka.TopicPartition,
 			Headers:        headers,
 			Value:          message.Value,
 		}, nil)
+
+		if counters.consumed%bc.config.ProducerQueueSize == 0 {
+			producer.Flush(bc.config.ProducerLingerMs)
+		}
+
 		if err != nil {
 			return counters, fmt.Errorf("failed to put message to producer: %v", err)
 		}