KAFKA-15951: MissingSourceTopicException should include topic names #15573

chickenchickenlove · 2024-03-21T14:28:59Z

This is minor changes!

Jira : https://issues.apache.org/jira/browse/KAFKA-15951
MissingSourceTopicException has field missingTopics to store missing topics.
The StreamsPartitionAssignor throws a MissingSourceTopicException. and it depends on result of repartitionTopics.missingSourceTopicExceptions().isEmpty(). thus, missingSourceTopicExceptions() must always contain an iterable MissingSourceTopicException, allowing for the aggregation and throwing of MissingSourceTopics.
Idea for improving more
- StreamsRebalanceListener can throw MissingSourceTopicException as well. however, StreamsRebalanceListener cannot get missing topic List because StreamsRebalanceListener depends on ErrorCode AssignorError.INCOMPLETE_SOURCE_TOPIC_METADATA.code()
- If StreamsRebalanceListener should include missing topic list when it throws MissingSourceTopicException as well, we can consider ThreadLocal to reach the target.
- Skeleton
  - Set ThreadLocal to StreamThread as member field.
  - Put missing topics to ThreadLocal.
  - StreamsRebalanceListener can get missing topics through ThreadLocal..

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

cadonna

Thanks for the PR @chickenchickenlove !

Here my comments!

cadonna · 2024-03-25T09:51:00Z

streams/src/main/java/org/apache/kafka/streams/errors/MissingSourceTopicException.java

+        this.missingTopics = new HashSet<>();
+    }
+
+    public MissingSourceTopicException(final String message, final Set<String> missingTopics) {
+        super(message);
+        this.missingTopics = missingTopics;
+    }
+
+    public Set<String> getMissingTopics() {
+        return this.missingTopics;


This class is part of the public API. That means, we cannot change it without a Kafka Improvement Proposal (KIP)[1].
I am not sure if adding the missing source topics to the exception makes too much sense, because the exception is caught in the StreamsPartitionAssignor [2] and transformed to a group assignment error (INCOMPLETE_SOURCE_TOPIC_METADATA). The missing source topics are not propagated to the point where the actual MissingSourceTopicException is thrown to the users, which is in the StreamsRebalanceListener [3]. To achieve this, a protocol change would be needed which I think it is not worth.
What we could do instead is log an error message with the missing source topics in RepartitionsTopics or StreamsPartitionAssignor. I slightly prefer the latter.

[1] https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Improvement+Proposals
[2]

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

Line 449 in d8dd068

} catch (final MissingSourceTopicException e) {

[3]

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

Line 58 in 4fe4cdc

throw new MissingSourceTopicException("One or more source topics were missing during rebalance");

Hi, @cadonna. Thanks for your comment and your time. 🙇‍♂️
It is very helpful for me, and i feel quite sorry to make you spend a lot of time for me.

After reading the KIP document, now i can tell public interface should be introduced carefully!
I have a comment to make new commit to apply your comment.

When you have some free time, could you take a look? it will be very helpful for me. 🙇‍♂️

Don't worry! We are also here to guide new contributors.

@cadonna thanks for your comment, really.
your comment encourages me really 🙇‍♂️

cadonna · 2024-03-25T10:04:04Z

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

+                    missingSourceTopics, subtopologyId, topologyName),
+                    missingSourceTopics),


Would it be possible to return a set of missing source topics from RepartitionTopics?

@cadonna, I have a question. ✋
May i use the package-private access modifier for this?
I think that package-private access modifier seems not be included to public interface, right? (https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Improvement+Proposals)

If so, i think that solution will be quite simple.

FYI, i write skeleton code below!
Does it make sense to you as well?

public class RepartitionTopics { ... // Add new field (private) private final Set<String> missingTopics = new HashSet(); ... public Set<String> topologiesWithMissingInputTopics() { ... } public Queue<StreamsException> missingSourceTopicExceptions() { ... } // Add new method (package-private) Set<String> getMissingTopics() { return this.missingTopics; } ... }

The public interface is defined as everything that shows up in the javadocs. Classes in a package whose name contains internals do not show up in the javadocs. Class RepartitionTopics is in package org.apache.kafka.streams.processor.internals. Thus, RepartitionTopics is not part of the public interface.
Regarding field missingTopics, there is already missingInputTopicsBySubtopology that includes all missing topics.

I really appreciate your kind explanation. 👍
I've understood it clearly, and thanks to you, I've established the correct direction for revision.

cadonna · 2024-03-25T10:12:34Z

...ams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java

@@ -523,7 +523,17 @@ private RepartitionTopics prepareRepartitionTopics(final Cluster metadata) {
        final boolean isMissingInputTopics = !repartitionTopics.missingSourceTopicExceptions().isEmpty();


Here you could then use the method on RepartitionTopcs I proposed in my other comment.

cadonna · 2024-03-25T10:34:12Z

...ams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java

@@ -523,7 +523,17 @@ private RepartitionTopics prepareRepartitionTopics(final Cluster metadata) {
        final boolean isMissingInputTopics = !repartitionTopics.missingSourceTopicExceptions().isEmpty();
        if (isMissingInputTopics) {
            if (!taskManager.topologyMetadata().hasNamedTopologies()) {
-                throw new MissingSourceTopicException("Missing source topics.");


In addition to throwing the exception you would also log the error.

Was Bruno meant was, that we need to add log.error(...) to log the error message before throwing the exception. Seems you did not add this yet?

Adding to a previous comment from Bruno, I am wondering if we should also change the error log in StreamsRebalanceListener to point out that the missing source topic names might be logged on a different instance?

The StreamsRebalanceListener is executed on every instance, but StreamsPartitionAssignor only on the group leader.

@mjsax thanks for your comments!

Was Bruno meant was, that we need to add log.error(...) to log the error message before throwing the exception. Seems you did not add this yet?

No, i created new commits. Please refer to images below 🙇‍♂️
In this Image, Bruno saidIn addition to throwing the exception you would also log the error.

and then, I have made these changes to reflect that comment to maintain readability.
you can see this commit.

Adding to a previous comment from Bruno, I am wondering if we should also change the error log in StreamsRebalanceListener to point out that the missing source topic names might be logged on a different instance?

The StreamsRebalanceListener is executed on every instance, but StreamsPartitionAssignor only on the group leader.

@mjsax , IMHO, currently with this PR alone, it is not possible to refer to Missing Topicson StreamRebalanceListener.

To get Missing topics on StreamRebalanceListener as well, i wrote suggestion on this PR. Please refer to Idea for improving more on this PR description. I think it can be done without modifying public API.

Also, i will create some images for detail. Wait a sec, please.

I draw image above to describe `idea for improve`. If you want to read `MissingSourceTopics` in StreamRebalanceListener as well, this workaround seems like it could be a way to handle it.

As you know, ThreadLocal provides storage specific to each thread.

Both StreamThread, StreamRebalanceListener, StreamPartitionAssignor are included on internal package. that means, it is not public API.

StreamRebalanceListener has reference of StreamThread Already.

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

Line 622 in 4ccbf16

this.rebalanceListener = new StreamsRebalanceListener(time, taskManager, this, this.log, this.assignmentErrorCode);

From this, I believe one StreamThread has own StreamRebalanceListener instance. thsu, ThreadLocal is suitable workaround in this case, i believe.

Thus, all the things that we should do, are 3step.

Add ThreadLocal to field of StreamThread.

Add method that put missing source topics to ThreadLocal before throw MissingSourceTopicExceptions,

Add some codes on StreamRebalanceListener to get MissingSourceTopics from ThreadLocal.

Does it make sense to you?

Seems there is some misunderstanding. Sorry for causing confusion.

This current code is not producing a log message:

final String errorMsg = String.format("Missing source topics. %s", repartitionTopics.missingSourceTopics()); throw new MissingSourceTopicException(errorMsg);

The code should be something like:

final String errorMsg = String.format("Missing source topics. %s", repartitionTopics.missingSourceTopics()); log.error(errorMsg); throw new MissingSourceTopicException(errorMsg);

I did not propose to include the topic names... As Bruno already pointed out, it would require a protocol change what seems to be overkill. In StreamsRebalanceListener, we currently log

Received error code 1

Cf https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsRebalanceListener.java#L56

I would propose to actually change AssignorError to contain a proper String. Error code 1 does not mean anything to users. Additionally, it might be good to just change the error message of the log line and the exception to say something like: "To check which topics are missing, please look into the logs of the consumer group leader. Only the leaders knows and logs the name of the missing topics."

@mjsax Thanks for your kindful descriptions. it was very helpful for me 🙇‍♂️

1.This current code is not producing a log message:

I misunderstood, thanks for your time!!
I added log.error() through this commit.

I would propose to actually change AssignorError to contain a proper String.

I agree with you.
Here's how I've implemented it based on your suggestion. (commit)
I don't know all the meaning of AssignorError codes exactly, so I referred to the existing error logs message related with AssignorError codes to write the messages.

What do you think? Do you think i'm on right direction?
When you have some free time, Please take a look 🙇‍♂️.

chickenchickenlove · 2024-03-26T11:54:54Z

Gently ping, @cadonna !
I make new commit to apply your comments.
When you have some free time, please take a look. 🙇‍♂️

mjsax

Thanks for the PR. Made a pass.

mjsax · 2024-03-28T11:21:25Z

...ams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java

@@ -523,7 +523,17 @@ private RepartitionTopics prepareRepartitionTopics(final Cluster metadata) {
        final boolean isMissingInputTopics = !repartitionTopics.missingSourceTopicExceptions().isEmpty();
        if (isMissingInputTopics) {
            if (!taskManager.topologyMetadata().hasNamedTopologies()) {
-                throw new MissingSourceTopicException("Missing source topics.");


Was Bruno meant was, that we need to add log.error(...) to log the error message before throwing the exception. Seems you did not add this yet?

mjsax · 2024-03-28T11:30:55Z

...ams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java

@@ -523,7 +523,17 @@ private RepartitionTopics prepareRepartitionTopics(final Cluster metadata) {
        final boolean isMissingInputTopics = !repartitionTopics.missingSourceTopicExceptions().isEmpty();
        if (isMissingInputTopics) {
            if (!taskManager.topologyMetadata().hasNamedTopologies()) {
-                throw new MissingSourceTopicException("Missing source topics.");


Adding to a previous comment from Bruno, I am wondering if we should also change the error log in StreamsRebalanceListener to point out that the missing source topic names might be logged on a different instance?

The StreamsRebalanceListener is executed on every instance, but StreamsPartitionAssignor only on the group leader.

chickenchickenlove · 2024-03-28T13:00:52Z

Thanks for the PR. Made a pass.

Aye, i hope so. 👍

chickenchickenlove · 2024-04-10T08:38:55Z

Gently ping, @mjsax .
Would you please take a look when you have some free time? 🙇‍♂️

mjsax

@chickenchickenlove -- sorry for stalling to review this further.

Overall LGTM. A few suggested for the error code description.

...ams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/AssignorError.java

…nals/assignment/AssignorError.java Co-authored-by: Matthias J. Sax <mjsax@apache.org>

chickenchickenlove · 2024-05-02T10:59:11Z

@mjsax thanks for your suggestion. it is very suitable 👍
I commit your suggestion. when you have free time, please take a look 🙇‍♂️

mjsax · 2024-05-03T01:00:01Z

There is a checkstyle error:

[2024-05-02T11:03:05.489Z] > Task :streams:checkstyleMain

[2024-05-02T11:03:05.489Z] [ant:checkstyle] [ERROR] /home/jenkins/workspace/Kafka_kafka-pr_PR-15573/streams/src/main/java/org/apache/kafka/streams/processor/internals/RepartitionTopics.java:113:45: '{' is not preceded with whitespace. [WhitespaceAround]

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

…nals/RepartitionTopics.java

mjsax · 2024-05-03T01:01:01Z

I took the liberty to commit a fix to retrigger the build.

chickenchickenlove · 2024-05-03T01:24:52Z

Hi @mjsax !
Sorry, I missed to check checkStyle and thanks for your working instead of me 🙇‍♂️

mjsax · 2024-05-10T02:35:50Z

Thanks for the PR @chickenchickenlove! Merged to trunk.

chickenchickenlove · 2024-05-10T02:47:06Z

Thanks for your guideline.
It was very helpful to me 🙇‍♂️🙇‍♂️🙇‍♂️

…pache#15573) MissingSourceTopicException should contain the name of the missing topic. There is one corner case for which we don't have the topic name at hand, but we can log the topic name somewhere else. Reviewers: Bruno Cadonna <bruno@confluent.io>, Matthias J. Sax <matthias@confluent.io>

KAFKA-15951: MissingSourceTopicException should include topic names

76f8ffc

chickenchickenlove closed this Mar 22, 2024

chickenchickenlove reopened this Mar 22, 2024

cadonna reviewed Mar 25, 2024

View reviewed changes

KAFKA-15951: add interal API to RepartitionTopics.

3f16d28

chickenchickenlove force-pushed the kafka-15951 branch from b1650ea to 3f16d28 Compare March 26, 2024 11:50

chickenchickenlove requested a review from cadonna March 26, 2024 11:54

mjsax added the streams label Mar 28, 2024

mjsax reviewed Mar 28, 2024

View reviewed changes

chickenchickenlove requested a review from mjsax March 28, 2024 12:16

chickenchickenlove added 2 commits March 28, 2024 23:24

KAFKA-15951: remain missing source topics on their logs.

fb957db

KAFKA-15951: improve readability of AssignorError.

94b9764

mjsax reviewed May 2, 2024

View reviewed changes

chickenchickenlove and others added 4 commits May 2, 2024 19:56

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

f708934

…nals/assignment/AssignorError.java Co-authored-by: Matthias J. Sax <mjsax@apache.org>

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

b4c5eaf

…nals/assignment/AssignorError.java Co-authored-by: Matthias J. Sax <mjsax@apache.org>

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

042f1e7

…nals/assignment/AssignorError.java Co-authored-by: Matthias J. Sax <mjsax@apache.org>

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

2cdcac6

…nals/assignment/AssignorError.java Co-authored-by: Matthias J. Sax <mjsax@apache.org>

chickenchickenlove requested a review from mjsax May 2, 2024 10:59

mjsax reviewed May 3, 2024

View reviewed changes

streams/src/main/java/org/apache/kafka/streams/processor/internals/RepartitionTopics.java Outdated Show resolved Hide resolved

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

1489a8f

…nals/RepartitionTopics.java

mjsax merged commit ff6d01c into apache:trunk May 10, 2024
1 check failed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-15951: MissingSourceTopicException should include topic names #15573

KAFKA-15951: MissingSourceTopicException should include topic names #15573

chickenchickenlove commented Mar 21, 2024 •

edited

Loading

cadonna left a comment

cadonna Mar 25, 2024

chickenchickenlove Mar 25, 2024

cadonna Mar 26, 2024

chickenchickenlove Mar 26, 2024

cadonna Mar 25, 2024

chickenchickenlove Mar 25, 2024

chickenchickenlove Mar 25, 2024 •

edited

Loading

cadonna Mar 26, 2024 •

edited

Loading

chickenchickenlove Mar 26, 2024

cadonna Mar 25, 2024

cadonna Mar 25, 2024

mjsax Mar 28, 2024

mjsax Mar 28, 2024

chickenchickenlove Mar 28, 2024 •

edited

Loading

chickenchickenlove Mar 28, 2024 •

edited

Loading

chickenchickenlove Mar 28, 2024 •

edited

Loading

mjsax Mar 28, 2024

chickenchickenlove Mar 28, 2024 •

edited

Loading

chickenchickenlove commented Mar 26, 2024

mjsax left a comment

mjsax Mar 28, 2024

mjsax Mar 28, 2024

chickenchickenlove commented Mar 28, 2024

chickenchickenlove commented Apr 10, 2024

mjsax left a comment

chickenchickenlove commented May 2, 2024

mjsax commented May 3, 2024

mjsax commented May 3, 2024

chickenchickenlove commented May 3, 2024

mjsax commented May 10, 2024

chickenchickenlove commented May 10, 2024

		missingSourceTopics, subtopologyId, topologyName),
		missingSourceTopics),

		@@ -523,7 +523,17 @@ private RepartitionTopics prepareRepartitionTopics(final Cluster metadata) {
		final boolean isMissingInputTopics = !repartitionTopics.missingSourceTopicExceptions().isEmpty();

KAFKA-15951: MissingSourceTopicException should include topic names #15573

KAFKA-15951: MissingSourceTopicException should include topic names #15573

Conversation

chickenchickenlove commented Mar 21, 2024 • edited Loading

Committer Checklist (excluded from commit message)

cadonna left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chickenchickenlove Mar 25, 2024 • edited Loading

Choose a reason for hiding this comment

cadonna Mar 26, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chickenchickenlove Mar 28, 2024 • edited Loading

Choose a reason for hiding this comment

chickenchickenlove Mar 28, 2024 • edited Loading

Choose a reason for hiding this comment

chickenchickenlove Mar 28, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chickenchickenlove Mar 28, 2024 • edited Loading

Choose a reason for hiding this comment

chickenchickenlove commented Mar 26, 2024

mjsax left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chickenchickenlove commented Mar 28, 2024

chickenchickenlove commented Apr 10, 2024

mjsax left a comment

Choose a reason for hiding this comment

chickenchickenlove commented May 2, 2024

mjsax commented May 3, 2024

mjsax commented May 3, 2024

chickenchickenlove commented May 3, 2024

mjsax commented May 10, 2024

chickenchickenlove commented May 10, 2024

chickenchickenlove commented Mar 21, 2024 •

edited

Loading

chickenchickenlove Mar 25, 2024 •

edited

Loading

cadonna Mar 26, 2024 •

edited

Loading

chickenchickenlove Mar 28, 2024 •

edited

Loading

chickenchickenlove Mar 28, 2024 •

edited

Loading

chickenchickenlove Mar 28, 2024 •

edited

Loading

chickenchickenlove Mar 28, 2024 •

edited

Loading