MINOR: retry upon missing source topic by RaidenE1 · Pull Request #20284 · apache/kafka

RaidenE1 · 2025-07-31T21:04:52Z

Implements a timeout mechanism (using maxPollTimeMs) that waits for
missing source topics to be created before failing, instead of
immediately throwing exceptions in the new Streams protocol.
Additionally, throw TopologyException when partition count mismatch is
detected.

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

lucasbru

Mostly looking good to me. I left a few comments to cleanup/simplify the code.

lucasbru · 2025-08-01T12:30:51Z

    private volatile KafkaFutureImpl<Uuid> producerInstanceIdFuture = new KafkaFutureImpl<>();

+    // Missing source topic timeout tracking
+    private long firstMissingSourceTopicTime = -1L;


Maybe it would make things slighly more easy to read if we'd use
org.apache.kafka.common.utils.Timer for this?

Also, can we rename this to a more generic topicsReadyTimer? I think we may want to reuse the timer to also time out when internal topics are not created in time.

lucasbru · 2025-08-01T12:31:31Z

    private volatile KafkaFutureImpl<Uuid> restoreConsumerInstanceIdFuture = new KafkaFutureImpl<>();
    private volatile KafkaFutureImpl<Uuid> producerInstanceIdFuture = new KafkaFutureImpl<>();

+    // Missing source topic timeout tracking


If you describe a member, I'd use a javadoc comment. But this comment isn't adding anything on top of the variable name, so maybe we can drop it altogether?

lucasbru · 2025-08-01T12:33:24Z

+                handleMissingSourceTopicsWithTimeout(missingTopicsDetail);
+            } else {
+                // Reset timeout tracking when no missing source topics are reported
+                firstMissingSourceTopicTime = -1L;


I think if you use org.apache.kafka.common.utils.Timer and call reset here, you don't need the inline comment.

aliehsaeedii · 2025-08-05T12:15:58Z

+
+        // Advance time beyond max.poll.interval.ms (default is 300000ms) to trigger timeout
+        mockTime.sleep(300001);
+


Suggestion: advance time less than 5 min and check if 2nd call throws exception and also check the log message (if easy) and then next step advancing time beyond 5 min as you did!

aliehsaeedii · 2025-08-05T12:19:15Z

+
+        // First call should not throw exception (within timeout)
+        thread.runOnceWithoutProcessingThreads();
+


The same suggestion (below) here.

aliehsaeedii · 2025-08-05T12:29:07Z

@RaidenE1 Thank you, the PR looks good to me now. I had a suggestion, but it’s not necessary to address since you’re already checking the condition in a later test.

mjsax · 2025-08-05T22:35:37Z

+            log.error(errorMsg);
+
+            // Reset timer for next timeout cycle
+            topicsReadyTimer.updateAndReset(maxPollTimeMs);


Why do we need to update the timer? We throw MissingSourceTopicException below, and this should shut down the thread?

mjsax · 2025-08-05T22:38:16Z

+                    final String errorMsg = status.statusDetail();
                    log.error(errorMsg);
-                    throw new MissingSourceTopicException(errorMsg);
+                    throw new TopologyException(errorMsg);


It seems this case is newly added, but we did not add a new test for it?

…eleted (#20735) Integration tests expecting `MissingSourceTopicException` were timing out and taking 5+ minutes because [PR #20284](#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>

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

retry upon missing source topic

11a0e2c

github-actions Bot added triage PRs from the community streams labels Jul 31, 2025

frankvicky added the ci-approved label Aug 1, 2025

lucasbru reviewed Aug 1, 2025

View reviewed changes

github-actions Bot removed the triage PRs from the community label Aug 2, 2025

RaidenE1 added 3 commits August 4, 2025 15:40

Use timer to setup the clock

bc685c6

spotless apply

283a9f0

spotlessapply

27ecd85

aliehsaeedii reviewed Aug 5, 2025

View reviewed changes

mjsax reviewed Aug 5, 2025

View reviewed changes

RaidenE1 added 2 commits August 6, 2025 15:03

fix

3268321

update test

0bd46db

mjsax approved these changes Aug 6, 2025

View reviewed changes

mjsax merged commit 03190e4 into apache:trunk Aug 6, 2025
20 of 24 checks passed

lucasbru mentioned this pull request Sep 1, 2025

KAFKA-18185: remove internal.leave.group.on.close config #19400

Merged

RaidenE1 deleted the retry-upon-missing-source-topic branch September 18, 2025 13:39

RaidenE1 mentioned this pull request Oct 20, 2025

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

Merged

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

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

MINOR: retry upon missing source topic#20284

MINOR: retry upon missing source topic#20284
mjsax merged 6 commits intoapache:trunkfrom
RaidenE1:retry-upon-missing-source-topic

RaidenE1 commented Jul 31, 2025 •

edited by github-actions Bot

Loading

Uh oh!

lucasbru left a comment

Uh oh!

lucasbru Aug 1, 2025

Uh oh!

lucasbru Aug 1, 2025

Uh oh!

lucasbru Aug 1, 2025

Uh oh!

lucasbru Aug 1, 2025

Uh oh!

Uh oh!

Uh oh!

aliehsaeedii Aug 5, 2025 •

edited

Loading

Uh oh!

aliehsaeedii Aug 5, 2025 •

edited

Loading

Uh oh!

aliehsaeedii commented Aug 5, 2025

Uh oh!

mjsax Aug 5, 2025

Uh oh!

mjsax Aug 5, 2025

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants


		// Advance time beyond max.poll.interval.ms (default is 300000ms) to trigger timeout
		mockTime.sleep(300001);


		// First call should not throw exception (within timeout)
		thread.runOnceWithoutProcessingThreads();

Conversation

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

Uh oh!

Uh oh!

lucasbru left a comment

Choose a reason for hiding this comment

Uh oh!

lucasbru Aug 1, 2025

Choose a reason for hiding this comment

Uh oh!

lucasbru Aug 1, 2025

Choose a reason for hiding this comment

Uh oh!

lucasbru Aug 1, 2025

Choose a reason for hiding this comment

Uh oh!

lucasbru Aug 1, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

aliehsaeedii Aug 5, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

aliehsaeedii Aug 5, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

aliehsaeedii commented Aug 5, 2025

Uh oh!

mjsax Aug 5, 2025

Choose a reason for hiding this comment

Uh oh!

mjsax Aug 5, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

RaidenE1 commented Jul 31, 2025 •

edited by github-actions Bot

Loading

aliehsaeedii Aug 5, 2025 •

edited

Loading

aliehsaeedii Aug 5, 2025 •

edited

Loading