From 160d33c759d006109d8129082f7cb2ab1d1ddce6 Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Mon, 7 Sep 2026 11:33:50 +0800 Subject: [PATCH] Reduce Kafka publish latency for messages --- docker/tinode/Dockerfile | 6 +++--- docker/tinode/Dockerfile.build | 6 +++--- docker/tinode/config.template | 4 ++-- server/kafka_runtime.go | 5 +++++ server/kafka_workbench_audit_test.go | 11 +++++++++++ server/meetingim.conf | 1 + server/tinode.conf | 1 + 7 files changed, 26 insertions(+), 8 deletions(-) diff --git a/docker/tinode/Dockerfile b/docker/tinode/Dockerfile index 7de3560..637d6cc 100644 --- a/docker/tinode/Dockerfile +++ b/docker/tinode/Dockerfile @@ -75,6 +75,7 @@ ENV KAFKA_CONSUMER_GROUP=im-server ENV KAFKA_DIAL_TIMEOUT_MS=3000 ENV KAFKA_WRITE_TIMEOUT_MS=3000 ENV KAFKA_READ_TIMEOUT_MS=3000 +ENV KAFKA_BATCH_TIMEOUT_MS=10 ENV KAFKA_REQUIRED_ACKS=1 ENV KAFKA_ALLOW_AUTO_TOPIC_CREATION=false ENV KAFKA_TLS_ENABLED=false @@ -82,7 +83,7 @@ ENV KAFKA_SASL_ENABLED=false ENV KAFKA_SASL_MECHANISM=SCRAM-SHA-512 ENV KAFKA_SASL_USERNAME= ENV KAFKA_SASL_PASSWORD= -ENV KAFKA_MAX_RETRY_COUNT=5 +ENV KAFKA_MAX_RETRY_COUNT=3 ENV KAFKA_RETRY_BACKOFF_MS='[10000,30000,60000,300000,900000]' ENV KAFKA_DEAD_LETTER_RETENTION_SECONDS=7776000 ENV KAFKA_TOPIC_ADMIN_COMMAND=im.admin.command @@ -90,10 +91,9 @@ ENV KAFKA_TOPIC_ADMIN_COMMAND_RESULT=im.admin.command.result ENV KAFKA_TOPIC_CONFIG_EVENT=im.config.event ENV KAFKA_TOPIC_BUSINESS_EVENT=im.business.event ENV KAFKA_TOPIC_MESSAGE_PERSIST=im.message.persist +ENV KAFKA_TOPIC_GROUP_MESSAGE_PERSIST=im.group.message.persist ENV KAFKA_TOPIC_MESSAGE_PUSH=im.message.push ENV KAFKA_TOPIC_FILE_SECURITY=im.file.security -ENV KAFKA_TOPIC_FILE_LIFECYCLE=im.file.lifecycle -ENV KAFKA_TOPIC_WORKBENCH_EVENT=im.workbench.event ENV KAFKA_TOPIC_CACHE_INVALIDATE=im.cache.invalidate ENV KAFKA_TOPIC_AUDIT_EVENT=im.audit.event ENV KAFKA_TOPIC_DEAD_LETTER=im.dead.letter diff --git a/docker/tinode/Dockerfile.build b/docker/tinode/Dockerfile.build index 3c7a66a..a50589e 100644 --- a/docker/tinode/Dockerfile.build +++ b/docker/tinode/Dockerfile.build @@ -55,6 +55,7 @@ ENV KAFKA_CONSUMER_GROUP=im-server ENV KAFKA_DIAL_TIMEOUT_MS=3000 ENV KAFKA_WRITE_TIMEOUT_MS=3000 ENV KAFKA_READ_TIMEOUT_MS=3000 +ENV KAFKA_BATCH_TIMEOUT_MS=10 ENV KAFKA_REQUIRED_ACKS=1 ENV KAFKA_ALLOW_AUTO_TOPIC_CREATION=false ENV KAFKA_TLS_ENABLED=false @@ -62,7 +63,7 @@ ENV KAFKA_SASL_ENABLED=false ENV KAFKA_SASL_MECHANISM=SCRAM-SHA-512 ENV KAFKA_SASL_USERNAME= ENV KAFKA_SASL_PASSWORD= -ENV KAFKA_MAX_RETRY_COUNT=5 +ENV KAFKA_MAX_RETRY_COUNT=3 ENV KAFKA_RETRY_BACKOFF_MS='[10000,30000,60000,300000,900000]' ENV KAFKA_DEAD_LETTER_RETENTION_SECONDS=7776000 ENV KAFKA_TOPIC_ADMIN_COMMAND=im.admin.command @@ -70,10 +71,9 @@ ENV KAFKA_TOPIC_ADMIN_COMMAND_RESULT=im.admin.command.result ENV KAFKA_TOPIC_CONFIG_EVENT=im.config.event ENV KAFKA_TOPIC_BUSINESS_EVENT=im.business.event ENV KAFKA_TOPIC_MESSAGE_PERSIST=im.message.persist +ENV KAFKA_TOPIC_GROUP_MESSAGE_PERSIST=im.group.message.persist ENV KAFKA_TOPIC_MESSAGE_PUSH=im.message.push ENV KAFKA_TOPIC_FILE_SECURITY=im.file.security -ENV KAFKA_TOPIC_FILE_LIFECYCLE=im.file.lifecycle -ENV KAFKA_TOPIC_WORKBENCH_EVENT=im.workbench.event ENV KAFKA_TOPIC_CACHE_INVALIDATE=im.cache.invalidate ENV KAFKA_TOPIC_AUDIT_EVENT=im.audit.event ENV KAFKA_TOPIC_DEAD_LETTER=im.dead.letter diff --git a/docker/tinode/config.template b/docker/tinode/config.template index d5d3946..0f52852 100644 --- a/docker/tinode/config.template +++ b/docker/tinode/config.template @@ -125,6 +125,7 @@ "dial_timeout_ms": $KAFKA_DIAL_TIMEOUT_MS, "write_timeout_ms": $KAFKA_WRITE_TIMEOUT_MS, "read_timeout_ms": $KAFKA_READ_TIMEOUT_MS, + "batch_timeout_ms": $KAFKA_BATCH_TIMEOUT_MS, "required_acks": $KAFKA_REQUIRED_ACKS, "allow_auto_topic_creation": $KAFKA_ALLOW_AUTO_TOPIC_CREATION, "tls_enabled": $KAFKA_TLS_ENABLED, @@ -141,10 +142,9 @@ "config_event": "$KAFKA_TOPIC_CONFIG_EVENT", "business_event": "$KAFKA_TOPIC_BUSINESS_EVENT", "message_persist": "$KAFKA_TOPIC_MESSAGE_PERSIST", + "group_message_persist": "$KAFKA_TOPIC_GROUP_MESSAGE_PERSIST", "message_push": "$KAFKA_TOPIC_MESSAGE_PUSH", "file_security": "$KAFKA_TOPIC_FILE_SECURITY", - "file_lifecycle": "$KAFKA_TOPIC_FILE_LIFECYCLE", - "workbench_event": "$KAFKA_TOPIC_WORKBENCH_EVENT", "cache_invalidate": "$KAFKA_TOPIC_CACHE_INVALIDATE", "audit_event": "$KAFKA_TOPIC_AUDIT_EVENT", "dead_letter": "$KAFKA_TOPIC_DEAD_LETTER" diff --git a/server/kafka_runtime.go b/server/kafka_runtime.go index 190821e..69be8e4 100644 --- a/server/kafka_runtime.go +++ b/server/kafka_runtime.go @@ -39,6 +39,7 @@ type kafkaConfig struct { DialTimeoutMS int `json:"dial_timeout_ms"` WriteTimeoutMS int `json:"write_timeout_ms"` ReadTimeoutMS int `json:"read_timeout_ms"` + BatchTimeoutMS int `json:"batch_timeout_ms"` RequiredAcks int `json:"required_acks"` AllowAutoTopic bool `json:"allow_auto_topic_creation"` TLSEnabled bool `json:"tls_enabled"` @@ -131,6 +132,7 @@ func initKafkaRuntime(raw json.RawMessage) error { Transport: transport, RequiredAcks: kafka.RequiredAcks(cfg.RequiredAcks), AllowAutoTopicCreation: cfg.AllowAutoTopic, + BatchTimeout: time.Duration(cfg.BatchTimeoutMS) * time.Millisecond, WriteTimeout: time.Duration(cfg.WriteTimeoutMS) * time.Millisecond, ReadTimeout: time.Duration(cfg.ReadTimeoutMS) * time.Millisecond, MaxAttempts: cfg.MaxRetryCount, @@ -162,6 +164,9 @@ func normalizeKafkaConfig(cfg *kafkaConfig) { if cfg.ReadTimeoutMS <= 0 { cfg.ReadTimeoutMS = 3000 } + if cfg.BatchTimeoutMS <= 0 { + cfg.BatchTimeoutMS = 10 + } if cfg.RequiredAcks == 0 { cfg.RequiredAcks = int(kafka.RequireOne) } diff --git a/server/kafka_workbench_audit_test.go b/server/kafka_workbench_audit_test.go index 8b12397..2aba4ea 100644 --- a/server/kafka_workbench_audit_test.go +++ b/server/kafka_workbench_audit_test.go @@ -30,6 +30,17 @@ func TestKafkaTopicsKeepOnlyMessageInternalAsyncTopics(t *testing.T) { } } +func TestNormalizeKafkaConfigUsesLowLatencyBatchTimeout(t *testing.T) { + cfg := kafkaConfig{} + normalizeKafkaConfig(&cfg) + if cfg.BatchTimeoutMS != 10 { + t.Fatalf("unexpected kafka batch timeout: %d", cfg.BatchTimeoutMS) + } + if cfg.MaxRetryCount != 3 { + t.Fatalf("unexpected kafka max retry count: %d", cfg.MaxRetryCount) + } +} + func TestEnqueueKafkaAuditEventPublishesCrossSystemEvent(t *testing.T) { oldKafkaBus := kafkaBus oldPublisher := publishKafkaAuditEventEnvelope diff --git a/server/meetingim.conf b/server/meetingim.conf index 60711bc..b273726 100644 --- a/server/meetingim.conf +++ b/server/meetingim.conf @@ -114,6 +114,7 @@ "dial_timeout_ms": 3000, "write_timeout_ms": 3000, "read_timeout_ms": 3000, + "batch_timeout_ms": 10, "required_acks": 1, "allow_auto_topic_creation": false, "tls_enabled": false, diff --git a/server/tinode.conf b/server/tinode.conf index 60711bc..b273726 100644 --- a/server/tinode.conf +++ b/server/tinode.conf @@ -114,6 +114,7 @@ "dial_timeout_ms": 3000, "write_timeout_ms": 3000, "read_timeout_ms": 3000, + "batch_timeout_ms": 10, "required_acks": 1, "allow_auto_topic_creation": false, "tls_enabled": false,