You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "RaidenE1 (via GitHub)" <gi...@apache.org> on 2025/10/20 19:32:03 UTC

[PR] KAFKA-19521: Fix consumption and leaving group when source topic is deleted [kafka]

RaidenE1 opened a new pull request, #20735:
URL: https://github.com/apache/kafka/pull/20735

   Integration tests expecting `MissingSourceTopicException` were timing out and taking 5+ minutes because [PR #20284](https://github.com/apache/kafka/pull/20284) set the missing topics timeout to `max.poll.interval.ms` (default 300 seconds).
   
   This is inappropriate because:
   - When source topics don't exist, actual poll frequency is `poll.ms` (100ms), not `max.poll.interval.ms`
   - The exception should be raised much faster based on heartbeat frequency
   
   ## Solution
   
   Use `2 * heartbeatIntervalMs` (from broker via KIP-1071) as the timeout instead:
   - Ensures at least one heartbeat is sent before raising the exception
   - Falls back to `max.poll.interval.ms` for backward compatibility
   - Fixes slow integration tests
   
   ## Changes
   
   1. Track `heartbeatIntervalMs` in `StreamsRebalanceData` and update it on each heartbeat response
   2. Use `2 * heartbeatIntervalMs` as timeout in `StreamThread.handleMissingSourceTopicsWithTimeout()`
   3. Update `HandlingSourceTopicDeletionIntegrationTest` to test both "classic" and "streams" protocols
   


-- 
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-19521: Fix consumption and leaving group when source topic is deleted [kafka]

Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #20735:
URL: https://github.com/apache/kafka/pull/20735#discussion_r2445964154


##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceData.java:
##########
@@ -395,4 +398,14 @@ public List<StreamsGroupHeartbeatResponseData.Status> statuses() {
         return statuses.get();
     }
 
+    /** Updated whenever a heartbeat response is received from the broker. */
+    public void setHeartbeatIntervalMs(final int heartbeatIntervalMs) {
+        this.heartbeatIntervalMs.set(heartbeatIntervalMs);
+    }
+
+    /** Returns the heartbeat interval in milliseconds, or -1 if not yet set. */
+    public int getHeartbeatIntervalMs() {
+        return heartbeatIntervalMs.get();
+    }

Review Comment:
   The getter method's documentation states it returns -1 if not yet set, but the field is initialized to -1 via `AtomicInteger(-1)` on line 333. Consider documenting this initialization in the field-level comment or the setter to clarify when/how the value transitions from -1 to a valid heartbeat interval.



##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -1587,11 +1587,21 @@ public void handleStreamsRebalanceData() {
     }
 
     private void handleMissingSourceTopicsWithTimeout(final String missingTopicsDetail) {
+        // Determine the timeout: use 2 * heartbeatIntervalMs if available, otherwise fall back to maxPollTimeMs
+        long timeoutMs = maxPollTimeMs;
+        if (streamsRebalanceData.isPresent()) {
+            final int heartbeatIntervalMs = streamsRebalanceData.get().getHeartbeatIntervalMs();
+            if (heartbeatIntervalMs > 0) {
+                // Use 2 * heartbeatIntervalMs to ensure at least one more heartbeat is sent

Review Comment:
   The comment says 'at least one more heartbeat' but the calculation ensures at least two heartbeat intervals (2 * heartbeatIntervalMs). Consider updating the comment to clarify this ensures time for approximately two heartbeats, or explain why this duration is sufficient.
   ```suggestion
                   // Use 2 * heartbeatIntervalMs to allow time for approximately two heartbeats to be sent
   ```



-- 
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-19521: Fix consumption and leaving group when source topic is deleted [kafka]

Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20735:
URL: https://github.com/apache/kafka/pull/20735#discussion_r2449109105


##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -1587,11 +1587,15 @@ public void handleStreamsRebalanceData() {
     }
 
     private void handleMissingSourceTopicsWithTimeout(final String missingTopicsDetail) {
+        // Determine the timeout: use 2 * heartbeatIntervalMs
+        final int heartbeatIntervalMs = streamsRebalanceData.get().getHeartbeatIntervalMs();
+        final long timeoutMs = 2 * heartbeatIntervalMs;

Review Comment:
   Good suggestions by Copilot. I would apply this change.



-- 
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-19521: Fix consumption and leaving group when source topic is deleted [kafka]

Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru merged PR #20735:
URL: https://github.com/apache/kafka/pull/20735


-- 
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-19521: Fix consumption and leaving group when source topic is deleted [kafka]

Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #20735:
URL: https://github.com/apache/kafka/pull/20735#discussion_r2447007929


##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -1587,11 +1587,15 @@ public void handleStreamsRebalanceData() {
     }
 
     private void handleMissingSourceTopicsWithTimeout(final String missingTopicsDetail) {
+        // Determine the timeout: use 2 * heartbeatIntervalMs
+        final int heartbeatIntervalMs = streamsRebalanceData.get().getHeartbeatIntervalMs();
+        final long timeoutMs = 2 * heartbeatIntervalMs;

Review Comment:
   Potential integer overflow when multiplying `heartbeatIntervalMs` by 2. If `heartbeatIntervalMs` is close to `Integer.MAX_VALUE`, the multiplication will overflow before being promoted to `long`. Cast `heartbeatIntervalMs` to `long` before multiplication: `final long timeoutMs = 2L * heartbeatIntervalMs;`
   ```suggestion
           final long timeoutMs = 2L * heartbeatIntervalMs;
   ```



##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -1587,11 +1587,15 @@ public void handleStreamsRebalanceData() {
     }
 
     private void handleMissingSourceTopicsWithTimeout(final String missingTopicsDetail) {
+        // Determine the timeout: use 2 * heartbeatIntervalMs
+        final int heartbeatIntervalMs = streamsRebalanceData.get().getHeartbeatIntervalMs();
+        final long timeoutMs = 2 * heartbeatIntervalMs;
+
         // Start timeout tracking on first encounter with missing topics
         if (topicsReadyTimer == null) {
-            topicsReadyTimer = time.timer(maxPollTimeMs);
+            topicsReadyTimer = time.timer(timeoutMs);

Review Comment:
   Missing validation for `heartbeatIntervalMs` being -1 (its default value when not yet set by a heartbeat response). If no heartbeat has been received, this will result in `timeoutMs = -2`, causing incorrect timer behavior. Add a check to fall back to `maxPollTimeMs` when `heartbeatIntervalMs == -1`.



##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceDataTest.java:
##########
@@ -437,4 +437,44 @@ public void streamsRebalanceDataShouldBeConstructedWithEmptyStatuses() {
         assertTrue(streamsRebalanceData.statuses().isEmpty());
     }
 
+    @Test
+    public void streamsRebalanceDataShouldBeConstructedWithHeartbeatIntervalMsSetToMinusOne() {
+        final UUID processId = UUID.randomUUID();
+        final Optional<StreamsRebalanceData.HostInfo> endpoint = Optional.of(new
+                StreamsRebalanceData.HostInfo("localhost", 9090));
+        final Map<String, StreamsRebalanceData.Subtopology> subtopologies = Map.of();
+        final Map<String, String> clientTags = Map.of("clientTag1",
+                "clientTagValue1");
+        final StreamsRebalanceData streamsRebalanceData = new StreamsRebalanceData(
+                processId,
+                endpoint,
+                subtopologies,
+                clientTags
+        );
+
+        assertEquals(-1, streamsRebalanceData.getHeartbeatIntervalMs());
+    }
+
+    @Test
+    public void streamsRebalanceDataShouldBeAbleToUpdateHeartbeatIntervalMs() {
+        final UUID processId = UUID.randomUUID();
+        final Optional<StreamsRebalanceData.HostInfo> endpoint = Optional.of(new
+                StreamsRebalanceData.HostInfo("localhost", 9090));
+        final Map<String, StreamsRebalanceData.Subtopology> subtopologies = Map.of();
+        final Map<String, String> clientTags = Map.of("clientTag1",
+                "clientTagValue1");
+        final StreamsRebalanceData streamsRebalanceData = new StreamsRebalanceData(
+                processId,
+                endpoint,
+                subtopologies,
+                clientTags
+        );
+
+        streamsRebalanceData.setHeartbeatIntervalMs(1000);
+        assertEquals(1000, streamsRebalanceData.getHeartbeatIntervalMs());
+    }
+
+
+
+

Review Comment:
   Remove unnecessary blank lines at the end of the file. There are three consecutive blank lines before the closing brace.
   ```suggestion
   
   ```



-- 
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-19521: Fix consumption and leaving group when source topic is deleted [kafka]

Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #20735:
URL: https://github.com/apache/kafka/pull/20735#discussion_r2445969927


##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -1587,11 +1587,21 @@ public void handleStreamsRebalanceData() {
     }
 
     private void handleMissingSourceTopicsWithTimeout(final String missingTopicsDetail) {
+        // Determine the timeout: use 2 * heartbeatIntervalMs if available, otherwise fall back to maxPollTimeMs

Review Comment:
   I think heartbeatIntervalMs must be defined at this stage. I would simplify the code, there is no need for some default.



##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -1587,11 +1587,21 @@ public void handleStreamsRebalanceData() {
     }
 
     private void handleMissingSourceTopicsWithTimeout(final String missingTopicsDetail) {
+        // Determine the timeout: use 2 * heartbeatIntervalMs if available, otherwise fall back to maxPollTimeMs
+        long timeoutMs = maxPollTimeMs;
+        if (streamsRebalanceData.isPresent()) {

Review Comment:
   isn't this only called when streamsRebalanceData is present?



##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceData.java:
##########
@@ -395,4 +398,14 @@ public List<StreamsGroupHeartbeatResponseData.Status> statuses() {
         return statuses.get();
     }
 
+    /** Updated whenever a heartbeat response is received from the broker. */
+    public void setHeartbeatIntervalMs(final int heartbeatIntervalMs) {
+        this.heartbeatIntervalMs.set(heartbeatIntervalMs);
+    }
+
+    /** Returns the heartbeat interval in milliseconds, or -1 if not yet set. */
+    public int getHeartbeatIntervalMs() {

Review Comment:
   We don't use the `get` prefix here.



-- 
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-19521: Fix consumption and leaving group when source topic is deleted [kafka]

Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20735:
URL: https://github.com/apache/kafka/pull/20735#discussion_r2449106870


##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -1587,11 +1587,15 @@ public void handleStreamsRebalanceData() {
     }
 
     private void handleMissingSourceTopicsWithTimeout(final String missingTopicsDetail) {
+        // Determine the timeout: use 2 * heartbeatIntervalMs

Review Comment:
   This comment just says, what the code does (-> `timeoutMs = 2 * heartbeatIntervalMs`) -- it does not add any knowledge. -- Either remove it, or say _why_ we need to use `2* heartbeatIntervalMs` as timeout



-- 
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