客户端 SDK 开发指南

# https://kafka.apache.org/documentation/#producerconfigs
# 批量大小，一次性最多积累 1MiB 的数据。
batch.size=1048576
# 攒批时间，对于 Producer，如果在指定的时间内没有达到攒批的上限，也会触发发送。这个时间代表了发送的最大延迟。
linger.ms=1000

# https://kafka.apache.org/21/documentation.html#producerconfigs
# 最大重试次数
retries=2147483647

# 初始退避延迟时间
retry.backoff.ms=100
# 最大退避延迟时间
retry.backoff.max.ms=1000
# 每次发送请求的 RPC 超时时间
request.timeout.ms=30000
# 整个发送调用整体超时时间，超过这个时间后即不再重试，返回调用侧失败异常。
delivery.timeout.ms=120000
# 批量大小，一次性最多积累 1MiB 的数据。
batch.size=1048576
# 攒批时间，对于 Producer，如果在指定的时间内没有达到攒批的上限，也会触发发送。这个时间代表了发送的最大延迟。
linger.ms=1000

package main

import (
	"time"

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

func main() {
	client, err := kgo.NewClient(
		kgo.SeedBrokers("localhost:9092"),
		// 对应 Kafka 配置中的 linger.ms，例如 1000 毫秒。
		kgo.ProducerLinger(1*time.Second),
	)
	if err != nil {
		panic(err)
	}
	defer client.Close()
}

// https://pkg.go.dev/github.com/segmentio/kafka-go#Writer
// 推荐配置示例：
w := &kafka.Writer{
	// 消息发送最大重试次数。
	// 参考 Apache Kafka Java Client 的 retries 参数，设置为 Integer.MAX_VALUE。
	MaxAttempts: 2147483647,
	// 触发一次发送的最大消息条数。
	BatchSize: 4096,
	// 需要的分区副本确认数。
	RequiredAcks: kafka.RequireAll,
}

package main

import (
	"time"

	"github.com/IBM/sarama"
)

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

	// 触发一次 flush 的字节数阈值。
	config.Producer.Flush.Bytes = 1048576
	// 触发一次 flush 的消息条数阈值。
	config.Producer.Flush.Messages = 4096
	// flush 的最大等待时间，等同于 Kafka Java Producer 的 linger.ms。
	config.Producer.Flush.Frequency = 1 * time.Second

	// 消息发送最大重试次数（默认 3），等同于 Kafka Java Producer 的 retries。
	config.Producer.Retry.Max = 2147483647
	// 重试之间的退避等待时间（默认 100ms），等同于 Kafka Java Producer 的 retry.backoff.ms。
	config.Producer.Retry.Backoff = 1 * time.Second

	// TODO: 使用 config 创建 producer，例如 sarama.NewSyncProducer(brokers, config)
	_ = config
}

# https://www.elastic.co/docs/reference/logstash/plugins/plugins-outputs-kafka
output {
  kafka {
    # 其他连接配置 ...
    # 批量大小（字节），一次最多可累积 1MiB 的数据。
    batch_size => 1048576
    # 批量等待时间（毫秒）。如果在此时间内未达到批量限制，仍会触发发送。
    linger_ms => 1000
    # 生产者要求 Leader 在确认请求完成之前已收到的确认数。
    acks => "all"
    # 生产者生成的所有数据的压缩类型。
    compression_type => "lz4"
    # 重试失败的生产请求之前的等待时间。
    retry_backoff_ms => 1000
  }
}

# https://www.elastic.co/docs/reference/beats/filebeat/kafka-output
# Leader 选举期间的重试退避等待时间。
retry.backoff: 3s
# 批量发送 Kafka 请求的等待时间。
bulk_flush_frequency: 1s

# https://github.com/fluent/fluent-plugin-kafka
<match **>
  @type kafka2
  # 其他连接配置 ...

  <buffer>
    # 使用默认配置
  </buffer>
</match>

# https://docs.fluentbit.io/manual/pipeline/outputs/kafka
pipeline:
  outputs:
    - name: kafka
      match: '*'
      # 其他连接配置 ...
      # 批量等待时间（毫秒）。
      rdkafka.queue.buffering.max.ms: 1000
      # 压缩类型。
      rdkafka.compression.type: lz4

# https://vector.dev/docs/reference/configuration/sinks/kafka/
[sinks.my_kafka]
type = "kafka"
inputs = ["my-source-or-transform-id"]
# 其他连接配置 ...
# 生产者生成的所有数据的压缩类型。
compression = "lz4"

# 批量等待时间通过 librdkafka 选项控制。
[sinks.my_kafka.librdkafka_options]
"queue.buffering.max.ms" = "1000"