Client SDK Guide

# https://kafka.apache.org/documentation/#producerconfigs
# Batch size, up to 1MiB of data can be accumulated at a time.
batch.size=1048576
# Batch linger time. If the batch limit is not reached within this time, sending is triggered anyway. This represents the maximum send delay.
linger.ms=1000

# https://kafka.apache.org/21/documentation.html#producerconfigs
# Maximum retry count
retries=2147483647

# Initial backoff delay time
retry.backoff.ms=100
# Maximum backoff delay time
retry.backoff.max.ms=1000
# RPC timeout for each send request
request.timeout.ms=30000
# Total timeout for the entire send call. No further retries after this time; an exception is returned to the caller.
delivery.timeout.ms=120000
# Batch size, up to 1MiB of data can be accumulated at a time.
batch.size=1048576
# Batch linger time. If the batch limit is not reached within this time, sending is triggered anyway. This represents the maximum send delay.
linger.ms=1000

package main

import (
	"time"

	"github.com/twmb/franz-go/pkg/kgo"
)

func main() {
	client, err := kgo.NewClient(
		kgo.SeedBrokers("localhost:9092"),
		// Equivalent to the linger.ms configuration in the Java client, set to 1 second.
		kgo.ProducerLinger(1*time.Second),
	)
	if err != nil {
		panic(err)
	}
	defer client.Close()
}

// https://pkg.go.dev/github.com/segmentio/kafka-go#Writer
// Configure kafka.Writer fields; values align with Apache Kafka Java producer defaults.
writer := &kafka.Writer{
	// Maximum number of attempts to deliver a message.
	// Aligned with the Apache Kafka Java Client retries parameter (Integer.MAX_VALUE).
	MaxAttempts: 2147483647,
	// Maximum number of messages to trigger a flush.
	BatchSize: 4096,
	// Number of acknowledges from partition replicas required before responding.
	RequiredAcks: kafka.RequireAll,
}

package main

import (
	"time"

	"github.com/IBM/sarama"
)

func main() {
	// https://pkg.go.dev/github.com/IBM/sarama#Config
	config := sarama.NewConfig()

	// Byte threshold to trigger a flush.
	config.Producer.Flush.Bytes = 1048576
	// Message count threshold to trigger a flush.
	config.Producer.Flush.Messages = 4096
	// Maximum flush wait time, equivalent to the Kafka Java producer's linger.ms.
	config.Producer.Flush.Frequency = 1 * time.Second

	// Maximum retry count (default 3), equivalent to the Kafka Java producer's retries.
	config.Producer.Retry.Max = 2147483647
	// Backoff wait time between retries (default 100ms), equivalent to the Kafka Java producer's retry.backoff.ms.
	config.Producer.Retry.Backoff = 1 * time.Second

	// TODO: use config to create a producer, e.g. sarama.NewSyncProducer(brokers, config)
	_ = config
}

# https://www.elastic.co/docs/reference/logstash/plugins/plugins-outputs-kafka
output {
  kafka {
    # other connection configs ...
    # Batch size in bytes, up to 1MiB of data can be accumulated at a time.
    batch_size => 1048576
    # Batch linger time in ms. If the batch limit is not reached within this time, sending is triggered anyway.
    linger_ms => 1000
    # The number of acknowledgments the producer requires the leader to have received.
    acks => "all"
    # Compression type for all data generated by the producer.
    compression_type => "lz4"
    # The amount of time to wait before attempting to retry a failed produce request.
    retry_backoff_ms => 1000
  }
}

# https://www.elastic.co/docs/reference/beats/filebeat/kafka-output
# Retry backoff wait time during leader elections.
retry.backoff: 3s
# Wait time before sending bulk Kafka requests.
bulk_flush_frequency: 1s

# https://github.com/fluent/fluent-plugin-kafka
<match **>
  @type kafka2
  # other connection configs ...

  <buffer>
    # use default configs
  </buffer>
</match>

# https://docs.fluentbit.io/manual/pipeline/outputs/kafka
pipeline:
  outputs:
    - name: kafka
      match: '*'
      # other connection configs ...
      # Batch linger time in ms.
      rdkafka.queue.buffering.max.ms: 1000
      # Compression type.
      rdkafka.compression.type: lz4

# https://vector.dev/docs/reference/configuration/sinks/kafka/
[sinks.my_kafka]
type = "kafka"
inputs = ["my-source-or-transform-id"]
# other connection configs ...
# Compression type for all data generated by the producer.
compression = "lz4"

# Batch linger time is controlled via librdkafka options.
[sinks.my_kafka.librdkafka_options]
"queue.buffering.max.ms" = "1000"