vvcephei commented on a change in pull request #9446:
URL: https://github.com/apache/kafka/pull/9446#discussion_r506567159
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsRebalanceListener.java
##########
@@ -52,8 +53,11 @@ public void onPartitionsAssigned(final
Collection<TopicPartition> partitions) {
// NB: all task management is already handled by:
//
org.apache.kafka.streams.processor.internals.StreamsPartitionAssignor.onAssignment
if (assignmentErrorCode.get() ==
AssignorError.INCOMPLETE_SOURCE_TOPIC_METADATA.code()) {
- log.error("Received error code {}", assignmentErrorCode.get());
+ log.error("Received error code {}",
AssignorError.INCOMPLETE_SOURCE_TOPIC_METADATA);
throw new MissingSourceTopicException("One or more source topics
were missing during rebalance");
+ } else if (assignmentErrorCode.get() ==
AssignorError.ASSIGNMENT_ERROR.code()) {
+ log.error("Received error code {}",
AssignorError.ASSIGNMENT_ERROR);
+ throw new TaskAssignmentException("Hit an unexpected exception
during task assignment phase of rebalance");
}
Review comment:
Hey @ableegoldman , I just looked into this, and it appears that this is
the only place that the error code would be read and thrown as an exception to
kill the thread, right?
If so, then it looks like older clients are looking specifically for the
error code to be `== AssignorError.INCOMPLETE_SOURCE_TOPIC_METADATA.code()`,
and they'd interpret any other code as "looks good" and proceed with
PARTITIONS_ASSIGNED.
Maybe there's nothing we can do about it now, but perhaps we should put in a
block to future-proof this code by adding an `else if
(assignmentErrorCode.get() != 0) { throw new TaskAssignmentException("Unknown
error code: "+assignmentErrorCode.get()) }` ?
Or is this already handled in some other way I'm not seeing?
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java
##########
@@ -409,9 +396,18 @@ public GroupAssignment assign(final Cluster metadata,
final GroupSubscription gr
minSupportedMetadataVersion,
versionProbing,
probingRebalanceNeeded
- );
+ );
- return new GroupAssignment(assignment);
+ return new GroupAssignment(assignment);
+ } catch (final MissingSourceTopicException e) {
+ return new GroupAssignment(
+ errorAssignment(clientMetadataMap,
AssignorError.INCOMPLETE_SOURCE_TOPIC_METADATA.code())
+ );
+ } catch (final TaskAssignmentException e) {
+ return new GroupAssignment(
+ errorAssignment(clientMetadataMap,
AssignorError.ASSIGNMENT_ERROR.code())
Review comment:
Following on my other comment, should we gate this on the
usedMetadataVersion so that we'd return a `INCOMPLETE_SOURCE_TOPIC_METADATA`
for versions 7 and below, to ensure they'll be properly handled as exceptions
on the member side?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
