KAFKA-19521: Fix consumption and leaving group when source topic is deleted #20735

RaidenE1 · 2025-10-20T19:32:00Z

Integration tests expecting MissingSourceTopicException were timing
out and taking 5+ minutes because PR
#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

Track heartbeatIntervalMs in StreamsRebalanceData and update it
on each heartbeat response
Use 2 * heartbeatIntervalMs as timeout in
StreamThread.handleMissingSourceTopicsWithTimeout()
Update HandlingSourceTopicDeletionIntegrationTest to test both
"classic" and "streams" protocols
Update StreamsRebalanceDataTest,
StreamsGroupHeartbeatRequestManagerTest and StreamThreadTest to test
StreamsRebalanceData and its behaviour

Reviewers: Lucas Brutschy lbrutschy@confluent.io, Matthias J. Sax
matthias@confluent.io

Copilot

Pull Request Overview

This PR fixes a timeout issue in integration tests where missing source topics were causing 5+ minute delays. The fix changes the timeout calculation from using max.poll.interval.ms (300s default) to 2 * heartbeatIntervalMs, which better reflects the actual polling behavior and ensures faster exception reporting.

Key Changes:

Modified timeout calculation to use heartbeat interval instead of max poll interval
Added heartbeat interval tracking in StreamsRebalanceData
Extended integration tests to cover both "classic" and "streams" group protocols

Reviewed Changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 2 comments.

File	Description
StreamThread.java	Updated timeout logic to use `2 * heartbeatIntervalMs` when available, falling back to `maxPollTimeMs`
StreamsRebalanceData.java	Added `heartbeatIntervalMs` field with getter/setter methods for tracking heartbeat interval
StreamsGroupHeartbeatRequestManager.java	Updated to set heartbeat interval in `StreamsRebalanceData` on successful heartbeat response
HandlingSourceTopicDeletionIntegrationTest.java	Converted test to parameterized test supporting both "classic" and "streams" protocols

_{Tip: Customize your code reviews with copilot-instructions.md. Create the file or learn how to get started.}

Copilot · 2025-10-20T19:51:24Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceData.java

+    /** 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();
+    }


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.

Copilot · 2025-10-20T19:51:25Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

+        if (streamsRebalanceData.isPresent()) {
+            final int heartbeatIntervalMs = streamsRebalanceData.get().getHeartbeatIntervalMs();
+            if (heartbeatIntervalMs > 0) {
+                // Use 2 * heartbeatIntervalMs to ensure at least one more heartbeat is sent


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.

Suggested change

// Use 2 * heartbeatIntervalMs to ensure at least one more heartbeat is sent

// Use 2 * heartbeatIntervalMs to allow time for approximately two heartbeats to be sent

lucasbru

You need to update

testStreamsProtocolRunOnceWithProcessingThreadsMissingSourceTopic
testStreamsProtocolMissingSourceTopicRecovery

Right? They still mention max.poll.interval

Also StreamsRebalanceDataTest probably needs an update and so does StreamsGroupHeartbeatRequestManagerTest

Can you please run the integration test 100 times in a row to ensure it's not flaky anymore?

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceData.java

Copilot

Pull Request Overview

Copilot reviewed 7 out of 7 changed files in this pull request and generated 3 comments.

_{Tip: Customize your code reviews with copilot-instructions.md. Create the file or learn how to get started.}

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

Copilot · 2025-10-21T07:24:26Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

+        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);


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

mjsax · 2025-10-21T17:15:02Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

    }

    private void handleMissingSourceTopicsWithTimeout(final String missingTopicsDetail) {
+        // Determine the timeout: use 2 * heartbeatIntervalMs


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

…nals/StreamThread.java Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

…als/StreamsRebalanceDataTest.java Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

lucasbru

LGTM, thanks!

…eleted (apache#20735) Integration tests expecting `MissingSourceTopicException` were timing out and taking 5+ minutes because [PR apache#20284](apache#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 4. Update `StreamsRebalanceDataTest`, `StreamsGroupHeartbeatRequestManagerTest` and `StreamThreadTest` to test `StreamsRebalanceData` and its behaviour Reviewers: Lucas Brutschy <lbrutschy@confluent.io>, Matthias J. Sax <matthias@confluent.io> --------- Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

resolve

9d02fae

github-actions bot added triage PRs from the community streams consumer clients small Small PRs labels Oct 20, 2025

lucasbru self-assigned this Oct 20, 2025

lucasbru requested review from Copilot and lucasbru October 20, 2025 19:50

lucasbru added the KIP-1071 PRs related to KIP-1071 label Oct 20, 2025

Copilot AI reviewed Oct 20, 2025

View reviewed changes

lucasbru added the ci-approved label Oct 20, 2025

lucasbru requested changes Oct 20, 2025

View reviewed changes

update

5dad457

github-actions bot removed the small Small PRs label Oct 21, 2025

fix

ef5121f

github-actions bot removed the triage PRs from the community label Oct 21, 2025

lucasbru requested a review from Copilot October 21, 2025 07:23

Copilot AI reviewed Oct 21, 2025

View reviewed changes

remove get prefix

be0d698

mjsax reviewed Oct 21, 2025

View reviewed changes

RaidenE1 and others added 3 commits October 21, 2025 16:07

Update streams/src/main/java/org/apache/kafka/streams/processor/inter…

ab344e7

…nals/StreamThread.java Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Update clients/src/test/java/org/apache/kafka/clients/consumer/intern…

7ec9ab5

…als/StreamsRebalanceDataTest.java Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

fix comment

87535c1

mjsax approved these changes Oct 22, 2025

View reviewed changes

lucasbru approved these changes Oct 22, 2025

View reviewed changes

lucasbru merged commit 2baad13 into apache:trunk Oct 22, 2025
24 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

KAFKA-19521: Fix consumption and leaving group when source topic is deleted #20735

KAFKA-19521: Fix consumption and leaving group when source topic is deleted #20735

Uh oh!

RaidenE1 commented Oct 20, 2025 •

edited by github-actions bot

Loading

Uh oh!

Copilot AI left a comment

Uh oh!

Copilot AI Oct 20, 2025

Uh oh!

Copilot AI Oct 20, 2025

Uh oh!

lucasbru left a comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Uh oh!

Uh oh!

Copilot AI Oct 21, 2025

Uh oh!

Uh oh!

mjsax Oct 21, 2025

Uh oh!

lucasbru left a comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

	// Use 2 * heartbeatIntervalMs to ensure at least one more heartbeat is sent
	// Use 2 * heartbeatIntervalMs to allow time for approximately two heartbeats to be sent

KAFKA-19521: Fix consumption and leaving group when source topic is deleted #20735

KAFKA-19521: Fix consumption and leaving group when source topic is deleted #20735

Uh oh!

Conversation

RaidenE1 commented Oct 20, 2025 • edited by github-actions bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Solution

Changes

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Reviewed Changes

Uh oh!

Copilot AI Oct 20, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Oct 20, 2025

Choose a reason for hiding this comment

Uh oh!

lucasbru left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Uh oh!

Uh oh!

Copilot AI Oct 21, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

mjsax Oct 21, 2025

Choose a reason for hiding this comment

Uh oh!

lucasbru left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

RaidenE1 commented Oct 20, 2025 •

edited by github-actions bot

Loading