From 7d6d279e9caa1296f495023fe6ce7f0f9429f617 Mon Sep 17 00:00:00 2001
From: Zhen Ye <chyezh@outlook.com>
Date: Mon, 22 Dec 2025 21:07:18 +0800
Subject: [PATCH] fix: set enable.auto.commit false to prevent from creating
 kafka consumer group (#46508)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

### **User description**
issue: #46507

we use the assign/unassign api to manage the consumer manually, the
commit operation will generate a new consumer group which is not what we
want. so we disable the auto commit to avoid it, also see:
https://github.com/confluentinc/confluent-kafka-python/issues/250#issuecomment-331377925


___

### **PR Type**
Bug fix


___

### **Description**
- Disable auto-commit in Kafka consumer configuration

- Prevents unwanted consumer group creation from manual offset
management

- Clarifies offset reset behavior with explanatory comments


___

### Diagram Walkthrough


```mermaid
flowchart LR
  A["Kafka Consumer Config"] --> B["Set enable.auto.commit to false"]
  B --> C["Prevent auto consumer group creation"]
  A --> D["Set auto.offset.reset to earliest"]
  D --> E["Handle deleted offsets gracefully"]
```



<details><summary><h3>File Walkthrough</h3></summary>

<table><thead><tr><th></th><th align="left">Relevant
files</th></tr></thead><tbody><tr><td><strong>Bug
fix</strong></td><td><table>
<tr>
  <td>
    <details>
<summary><strong>builder.go</strong><dd><code>Disable auto-commit and
add configuration comments</code>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
&nbsp; &nbsp; </dd></summary>
<hr>

pkg/streaming/walimpls/impls/kafka/builder.go

<ul><li>Added <code>enable.auto.commit</code> configuration set to
<code>false</code> to prevent <br>automatic consumer group creation<br>
<li> Added explanatory comments for both <code>auto.offset.reset</code>
and <br><code>enable.auto.commit</code> settings<br> <li> Clarifies that
manual assign/unassign API is used for consumer <br>management</ul>


</details>


  </td>
<td><a
href="https://github.com/milvus-io/milvus/pull/46508/files#diff-4b5635821fdc8b585d16c02d8a3b59079d8e667b2be43a073265112d72701add">+7/-0</a>&nbsp;
&nbsp; &nbsp; </td>

</tr>
</table></td></tr></tbody></table>

</details>

___



<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->

## Summary by CodeRabbit

## Bug Fixes
* Kafka consumer now reads from the earliest available messages and
auto-commit has been disabled to support manual offset management.

<sub>✏️ Tip: You can customize this high-level summary in your review
settings.</sub>

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Signed-off-by: chyezh <chyezh@outlook.com>
---
 pkg/streaming/walimpls/impls/kafka/builder.go | 7 +++++++
 1 file changed, 7 insertions(+)

diff --git a/pkg/streaming/walimpls/impls/kafka/builder.go b/pkg/streaming/walimpls/impls/kafka/builder.go
index bf1139641b..61c7673d6c 100644
--- a/pkg/streaming/walimpls/impls/kafka/builder.go
+++ b/pkg/streaming/walimpls/impls/kafka/builder.go
@@ -54,7 +54,14 @@ func (b *builderImpl) getConsumerConfig() kafka.ConfigMap {
 	config := &paramtable.Get().KafkaCfg
 	consumerConfig := getBasicConfig(config)
 	consumerConfig.SetKey("allow.auto.create.topics", true)
+	// may be the offset is already deleted by the retention policy of kafka.
+	// so we need to auto reset the offset to the earliest to consume reading.
 	consumerConfig.SetKey("auto.offset.reset", "earliest")
+	// we use the assign/unassign api to manage the consumer manually,
+	// the commit operation will generate a new consumer group which is not what we want.
+	// so we disable the auto commit to avoid it, also see:
+	// https://github.com/confluentinc/confluent-kafka-python/issues/250#issuecomment-331377925
+	consumerConfig.SetKey("enable.auto.commit", false)
 	for k, v := range config.ConsumerExtraConfig.GetValue() {
 		consumerConfig.SetKey(k, v)
 	}