You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "frankvicky (via GitHub)" <gi...@apache.org> on 2025/03/18 06:53:28 UTC
[PR] KAFKA-18991: FetcherThread should match leader epochs between fetch request and fetch state [kafka]
frankvicky opened a new pull request, #19223:
URL: https://github.com/apache/kafka/pull/19223
JIRA: KAFKA-18991
This PR fixes a potential issue where the `FetchResponse` returns `divergingEndOffsets` with an older leader epoch. This can lead to committed records being removed from the follower's log, potentially causing data loss.
In detail:
`processFetchRequest` gets the requested leader epoch of partition data by `topicPartition` and compares it with the leader epoch of the current fetch state. If they don't match, the response is ignored.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18991: FetcherThread should match leader epochs between fetch request and fetch state [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #19223:
URL: https://github.com/apache/kafka/pull/19223#issuecomment-2752123826
I try to cherry-pick https://github.com/apache/kafka/commit/4a8a0637e07734779b40ba9785842311144f922c to 4.0, but there are some conflicts. @jsancio do you have free cycle to file PR to cherry-pick it to 4.0?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18991: FetcherThread should match leader epochs between fetch request and fetch state [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #19223:
URL: https://github.com/apache/kafka/pull/19223#issuecomment-2752004577
> Could you provide a PR for the 3.9 and 4.0 branch as well?
should we backport KAFKA-18723 first?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18991: FetcherThread should match leader epochs between fetch request and fetch state [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #19223:
URL: https://github.com/apache/kafka/pull/19223#issuecomment-2752082932
> The fixed version for [KAFKA-18723](https://issues.apache.org/jira/browse/KAFKA-18723) already includes 3.9.1 and 4.0.1?
I grep the git history , and KAFKA-18723 is not in 4.0/3.9.
```
ubuntu@cncf-ubuntu-0:~/project/kafka$ git switch 4.0
Switched to branch '4.0'
Your branch is up to date with 'origin/4.0'.
ubuntu@cncf-ubuntu-0:~/project/kafka$ git log --grep='18723'
ubuntu@cncf-ubuntu-0:~/project/kafka$
ubuntu@cncf-ubuntu-0:~/project/kafka$ git switch 3.9
Switched to branch '3.9'
Your branch is up to date with 'origin/3.9'.
ubuntu@cncf-ubuntu-0:~/project/kafka$ git log --grep='18723'
ubuntu@cncf-ubuntu-0:~/project/kafka$
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18991: FetcherThread should match leader epochs between fetch request and fetch state [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on PR #19223:
URL: https://github.com/apache/kafka/pull/19223#issuecomment-2753131511
I have just tried to cherry-pick this patch, but as @chia7712 mentioned.
It seems that we should pick #18852 first.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18991: FetcherThread should match leader epochs between fetch request and fetch state [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao merged PR #19223:
URL: https://github.com/apache/kafka/pull/19223
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18991: FetcherThread should match leader epochs between fetch request and fetch state [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #19223:
URL: https://github.com/apache/kafka/pull/19223#discussion_r2008619688
##########
core/src/test/scala/unit/kafka/server/AbstractFetcherThreadTest.scala:
##########
@@ -1153,4 +1153,29 @@ class AbstractFetcherThreadTest {
assertTrue(fetcher.fetchState(unknownPartition).isEmpty)
}
-}
+ @Test
+ def testIgnoreFetchResponseWhenLeaderEpochChanged(): Unit = {
+ val newEpoch = 1
+ val initEpoch = 0
+
+ val partition = new TopicPartition("topic", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+ val replicaState = PartitionState(leaderEpoch = newEpoch)
+ fetcher.setReplicaState(partition, replicaState)
+ val initFetchState = initialFetchState(topicIds.get(partition.topic), 0L, leaderEpoch = newEpoch)
+ fetcher.addPartitions(Map(partition -> initFetchState))
+
+ val batch = mkBatch(baseOffset = 0L, leaderEpoch = 0, new SimpleRecord("a".getBytes))
+ val leaderState = PartitionState(Seq(batch), leaderEpoch = initEpoch, highWatermark = 1L)
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
Review Comment:
I think only L1173 is redundant.
Technically, these lines don't affect the test result if `initEpoch != newEpoch`.
But since these lines implement the mock mechanism of fetch, these mocks will return a non-empty `FetchResponse` as essential safeguards:
- If `initEpoch` is accidentally set equal to `newEpoch`, the response will be handled, and the leader epoch will be bumped as the leader epoch of fetchResponsethe test to fail.
- If `AbstractFetcherThread#L334` is accidentally removed, the response will be handled, and the leader epoch will be bumped as the leader epoch of fetchResponsethe test to fail.
Given that, we should keep these lines. WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18991: FetcherThread should match leader epochs between fetch request and fetch state [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #19223:
URL: https://github.com/apache/kafka/pull/19223#discussion_r2008619688
##########
core/src/test/scala/unit/kafka/server/AbstractFetcherThreadTest.scala:
##########
@@ -1153,4 +1153,29 @@ class AbstractFetcherThreadTest {
assertTrue(fetcher.fetchState(unknownPartition).isEmpty)
}
-}
+ @Test
+ def testIgnoreFetchResponseWhenLeaderEpochChanged(): Unit = {
+ val newEpoch = 1
+ val initEpoch = 0
+
+ val partition = new TopicPartition("topic", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+ val replicaState = PartitionState(leaderEpoch = newEpoch)
+ fetcher.setReplicaState(partition, replicaState)
+ val initFetchState = initialFetchState(topicIds.get(partition.topic), 0L, leaderEpoch = newEpoch)
+ fetcher.addPartitions(Map(partition -> initFetchState))
+
+ val batch = mkBatch(baseOffset = 0L, leaderEpoch = 0, new SimpleRecord("a".getBytes))
+ val leaderState = PartitionState(Seq(batch), leaderEpoch = initEpoch, highWatermark = 1L)
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
Review Comment:
I think only L1173 is redundant.
Technically, these lines don't affect the test result if `initEpoch != newEpoch`.
But since these lines implement the mock mechanism of fetch, they provide essential safeguards:
- If `initEpoch` is accidentally set equal to `newEpoch`, these mock lines will trigger specific logic that causes the test to fail.
- If `AbstractFetcherThread#L334` is accidentally removed, the mock will return an empty response, which will cause test assertion failures.
Given that, we should keep these lines. WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18991: FetcherThread should match leader epochs between fetch request and fetch state [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on PR #19223:
URL: https://github.com/apache/kafka/pull/19223#issuecomment-2740970168
Hi @junrao
Thanks for the review.
This is a tricky one to test.
I'm still researching how to create a condition that matches the patch's purpose.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18991: FetcherThread should match leader epochs between fetch request and fetch state [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #19223:
URL: https://github.com/apache/kafka/pull/19223#discussion_r2011192553
##########
core/src/test/scala/unit/kafka/server/AbstractFetcherThreadTest.scala:
##########
@@ -1153,4 +1153,28 @@ class AbstractFetcherThreadTest {
assertTrue(fetcher.fetchState(unknownPartition).isEmpty)
}
-}
+ @Test
+ def testIgnoreFetchResponseWhenLeaderEpochChanged(): Unit = {
+ val newEpoch = 1
+ val initEpoch = 0
+
+ val partition = new TopicPartition("topic", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+ val replicaState = PartitionState(leaderEpoch = newEpoch)
+ fetcher.setReplicaState(partition, replicaState)
+ val initFetchState = initialFetchState(topicIds.get(partition.topic), 0L, leaderEpoch = newEpoch)
+ fetcher.addPartitions(Map(partition -> initFetchState))
+
+ val batch = mkBatch(baseOffset = 0L, leaderEpoch = 0, new SimpleRecord("a".getBytes))
+ val leaderState = PartitionState(Seq(batch), leaderEpoch = initEpoch, highWatermark = 1L)
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+
+ val partitionData = Map(partition -> new FetchRequest.PartitionData(Uuid.randomUuid(), 0, 0, 1048576, Optional.of(initEpoch), Optional.of(0))).asJava
Review Comment:
Sure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18991: FetcherThread should match leader epochs between fetch request and fetch state [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19223:
URL: https://github.com/apache/kafka/pull/19223#discussion_r2006206448
##########
core/src/main/scala/kafka/server/AbstractFetcherThread.scala:
##########
@@ -333,11 +333,14 @@ abstract class AbstractFetcherThread(name: String,
responseData.foreachEntry { (topicPartition, partitionData) =>
Option(partitionStates.stateValue(topicPartition)).foreach { currentFetchState =>
// It's possible that a partition is removed and re-added or truncated when there is a pending fetch request.
- // In this case, we only want to process the fetch response if the partition state is ready for fetch and
- // the current offset is the same as the offset requested.
+ // In this case, we only want to process the fetch response if:
+ // - the partition state is ready for fetch
+ // - the current offset is the same as the offset requested
+ // - the current leader epoch is the same as the leader epoch requested
val fetchPartitionData = sessionPartitions.get(topicPartition)
if (fetchPartitionData != null &&
fetchPartitionData.fetchOffset == currentFetchState.fetchOffset &&
+ fetchPartitionData.currentLeaderEpoch.map[Boolean](_ == currentFetchState.currentLeaderEpoch).orElse(true) &&
Review Comment:
the line#368 can be streamlined to `currentFetchState.currentLeaderEpoch` due to this new condition.
```
val logAppendInfoOpt = processPartitionData(
topicPartition,
currentFetchState.fetchOffset,
fetchPartitionData.currentLeaderEpoch.orElse(currentFetchState.currentLeaderEpoch), // this line
partitionData
)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18991: FetcherThread should match leader epochs between fetch request and fetch state [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on PR #19223:
URL: https://github.com/apache/kafka/pull/19223#issuecomment-2752066965
> should we backport [KAFKA-18723](https://issues.apache.org/jira/browse/KAFKA-18723) first?
The fixed version for KAFKA-18723 already includes 3.9.1 and 4.0.1?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18991: FetcherThread should match leader epochs between fetch request and fetch state [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #19223:
URL: https://github.com/apache/kafka/pull/19223#discussion_r2008619688
##########
core/src/test/scala/unit/kafka/server/AbstractFetcherThreadTest.scala:
##########
@@ -1153,4 +1153,29 @@ class AbstractFetcherThreadTest {
assertTrue(fetcher.fetchState(unknownPartition).isEmpty)
}
-}
+ @Test
+ def testIgnoreFetchResponseWhenLeaderEpochChanged(): Unit = {
+ val newEpoch = 1
+ val initEpoch = 0
+
+ val partition = new TopicPartition("topic", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+ val replicaState = PartitionState(leaderEpoch = newEpoch)
+ fetcher.setReplicaState(partition, replicaState)
+ val initFetchState = initialFetchState(topicIds.get(partition.topic), 0L, leaderEpoch = newEpoch)
+ fetcher.addPartitions(Map(partition -> initFetchState))
+
+ val batch = mkBatch(baseOffset = 0L, leaderEpoch = 0, new SimpleRecord("a".getBytes))
+ val leaderState = PartitionState(Seq(batch), leaderEpoch = initEpoch, highWatermark = 1L)
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
Review Comment:
I think only L1173 is redundant.
Technically, these lines don't affect the test result if `initEpoch != newEpoch`.
But since these lines implement the mock mechanism of fetch, they provide essential safeguards:
- If `initEpoch` is accidentally set equal to `newEpoch`, these mock lines will trigger specific logic that causes the test to fail.
- If `AbstractFetcherThread#L334` is accidentally removed, the test case will fail; otherwise, it won't.
Given that, we should keep these lines. WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18991: FetcherThread should match leader epochs between fetch request and fetch state [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on PR #19223:
URL: https://github.com/apache/kafka/pull/19223#issuecomment-2751810808
@frankvicky : Could you provide a PR for the 3.9 and 4.0 branch as well?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18991: FetcherThread should match leader epochs between fetch request and fetch state [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #19223:
URL: https://github.com/apache/kafka/pull/19223#discussion_r2011073213
##########
core/src/test/scala/unit/kafka/server/AbstractFetcherThreadTest.scala:
##########
@@ -1153,4 +1153,28 @@ class AbstractFetcherThreadTest {
assertTrue(fetcher.fetchState(unknownPartition).isEmpty)
}
-}
+ @Test
+ def testIgnoreFetchResponseWhenLeaderEpochChanged(): Unit = {
+ val newEpoch = 1
+ val initEpoch = 0
+
+ val partition = new TopicPartition("topic", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+ val replicaState = PartitionState(leaderEpoch = newEpoch)
+ fetcher.setReplicaState(partition, replicaState)
+ val initFetchState = initialFetchState(topicIds.get(partition.topic), 0L, leaderEpoch = newEpoch)
+ fetcher.addPartitions(Map(partition -> initFetchState))
+
+ val batch = mkBatch(baseOffset = 0L, leaderEpoch = 0, new SimpleRecord("a".getBytes))
+ val leaderState = PartitionState(Seq(batch), leaderEpoch = initEpoch, highWatermark = 1L)
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+
+ val partitionData = Map(partition -> new FetchRequest.PartitionData(Uuid.randomUuid(), 0, 0, 1048576, Optional.of(initEpoch), Optional.of(0))).asJava
Review Comment:
Could we use `initEpoch` for `lastFetchedEpoch` too?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18991: FetcherThread should match leader epochs between fetch request and fetch state [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #19223:
URL: https://github.com/apache/kafka/pull/19223#discussion_r2007966429
##########
core/src/test/scala/unit/kafka/server/AbstractFetcherThreadTest.scala:
##########
@@ -1153,4 +1153,29 @@ class AbstractFetcherThreadTest {
assertTrue(fetcher.fetchState(unknownPartition).isEmpty)
}
-}
+ @Test
+ def testIgnoreFetchResponseWhenLeaderEpochChanged(): Unit = {
+ val newEpoch = 1
+ val initEpoch = 0
+
+ val partition = new TopicPartition("topic", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+ val replicaState = PartitionState(leaderEpoch = newEpoch)
+ fetcher.setReplicaState(partition, replicaState)
+ val initFetchState = initialFetchState(topicIds.get(partition.topic), 0L, leaderEpoch = newEpoch)
+ fetcher.addPartitions(Map(partition -> initFetchState))
+
+ val batch = mkBatch(baseOffset = 0L, leaderEpoch = 0, new SimpleRecord("a".getBytes))
+ val leaderState = PartitionState(Seq(batch), leaderEpoch = initEpoch, highWatermark = 1L)
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
Review Comment:
These three lines seem unneeded?
##########
core/src/test/scala/unit/kafka/server/AbstractFetcherThreadTest.scala:
##########
@@ -1153,4 +1153,29 @@ class AbstractFetcherThreadTest {
assertTrue(fetcher.fetchState(unknownPartition).isEmpty)
}
-}
+ @Test
+ def testIgnoreFetchResponseWhenLeaderEpochChanged(): Unit = {
+ val newEpoch = 1
+ val initEpoch = 0
+
+ val partition = new TopicPartition("topic", 0)
+ val mockLeaderEndpoint = new MockLeaderEndPoint(version = version)
+ val mockTierStateMachine = new MockTierStateMachine(mockLeaderEndpoint)
+ val fetcher = new MockFetcherThread(mockLeaderEndpoint, mockTierStateMachine)
+ val replicaState = PartitionState(leaderEpoch = newEpoch)
+ fetcher.setReplicaState(partition, replicaState)
+ val initFetchState = initialFetchState(topicIds.get(partition.topic), 0L, leaderEpoch = newEpoch)
+ fetcher.addPartitions(Map(partition -> initFetchState))
+
+ val batch = mkBatch(baseOffset = 0L, leaderEpoch = 0, new SimpleRecord("a".getBytes))
+ val leaderState = PartitionState(Seq(batch), leaderEpoch = initEpoch, highWatermark = 1L)
+ fetcher.mockLeader.setLeaderState(partition, leaderState)
+ fetcher.mockLeader.setReplicaPartitionStateCallback(fetcher.replicaPartitionState)
+
+ val partitionData = Map(partition -> new FetchRequest.PartitionData(Uuid.randomUuid(), 0, 0, 1048576, Optional.of(0), Optional.of(0))).asJava
Review Comment:
Could we use initEpoch instead of 0?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org