lucasbru commented on code in PR #22559:
URL: https://github.com/apache/kafka/pull/22559#discussion_r3458382945
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java:
##########
@@ -679,6 +679,11 @@ private void onErrorResponse(final
StreamsGroupHeartbeatResponse response, final
heartbeatRequestState.reset();
break;
+ case UNRELEASED_INSTANCE_ID:
+ logger.error("StreamsGroupHeartbeatRequest failed due to {}:
{}", error, errorMessage);
+ handleFatalFailure(error.exception(errorMessage));
+ break;
+
case UNSUPPORTED_VERSION:
Review Comment:
Now that static membership is enabled, FENCED_INSTANCE_ID becomes reachable
— GroupMetadataManager has a throwIfInstanceIdIsFenced overload for Streams. It
currently falls to the default branch with a generic 'unexpected error
FENCED_INSTANCE_ID' log. ConsumerHeartbeatRequestManager has an explicit case
with the message 'This is expected in the case that the member was removed from
the group by an admin client, and another member joined using the same group
instance id.' Worth adding the same here so operators get an actionable message.
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/KafkaStreamsCloseOptionsIntegrationTest.java:
##########
@@ -246,6 +251,48 @@ public void testCloseOptionsRemainInGroupStreamsProtocol()
throws Exception {
"Group should still have a member after REMAIN_IN_GROUP close
under Streams protocol");
}
+ @ParameterizedTest
+ @EnumSource(value = CloseOptions.GroupMembershipOperation.class, names =
{"DEFAULT", "REMAIN_IN_GROUP"})
+ public void testStaticMemberCloseUsesStaticLeaveEpochStreamsProtocol(
+ final CloseOptions.GroupMembershipOperation operation
+ ) throws Exception {
+ final int numStreamThreads = 2;
+ streamsConfig.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG,
numStreamThreads);
+ streamsConfig.put(StreamsConfig.GROUP_PROTOCOL_CONFIG,
GroupProtocol.STREAMS.name());
+
+ streams = new
KafkaStreams(setupTopologyWithoutIntermediateUserTopic(), streamsConfig);
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(streams);
+
IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(resultConsumerConfig,
OUTPUT_TOPIC, 10);
+
+ streams.close(CloseOptions.groupMembershipOperation(operation)
+ .withTimeout(Duration.ofSeconds(30)));
+
+ waitForStaticStreamsGroupMembersEpoch(
+ streamsConfig.getProperty(StreamsConfig.APPLICATION_ID_CONFIG),
+ GROUP_INSTANCE_ID,
+ StreamsGroupHeartbeatRequest.LEAVE_GROUP_STATIC_MEMBER_EPOCH,
+ numStreamThreads
+ );
+ }
+
+ @Test
+ public void testStaticMemberLeaveGroupStreamsProtocol() throws Exception {
+ streamsConfig.put(StreamsConfig.GROUP_PROTOCOL_CONFIG,
GroupProtocol.STREAMS.name());
+
+ streams = new
KafkaStreams(setupTopologyWithoutIntermediateUserTopic(), streamsConfig);
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(streams);
+
IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(resultConsumerConfig,
OUTPUT_TOPIC, 10);
+
+
streams.close(CloseOptions.groupMembershipOperation(CloseOptions.GroupMembershipOperation.LEAVE_GROUP)
+ .withTimeout(Duration.ofSeconds(30)));
+
+ waitForEmptyStreamGroup(
Review Comment:
Passing timeoutMs=0 to waitForEmptyStreamGroup means a single-shot check
with no retry window: TestUtils.retryOnExceptionWithTimeout sets
expectedEnd=now, calls the condition once, and if not met throws immediately.
streams.close() returning doesn't guarantee the coordinator has processed the
leave heartbeat yet. Under load this will be flaky. Same issue exists in
testCloseOptionsLeaveGroupStreamsProtocol and
testCloseOptionsDefaultStreamsProtocol in this class.
Should pass DEFAULT_MAX_WAIT_MS or a reasonable explicit timeout here.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java:
##########
@@ -679,6 +679,11 @@ private void onErrorResponse(final
StreamsGroupHeartbeatResponse response, final
heartbeatRequestState.reset();
break;
+ case UNRELEASED_INSTANCE_ID:
Review Comment:
Nit: the body here is byte-for-byte identical to the INVALID_REQUEST /
GROUP_MAX_SIZE_REACHED / STREAMS_INVALID_TOPOLOGY /
STREAMS_INVALID_TOPOLOGY_EPOCH / STREAMS_TOPOLOGY_FENCED fall-through group
(same logger.error call + handleFatalFailure(error.exception(errorMessage))).
UNSUPPORTED_VERSION is standalone because it substitutes
UNSUPPORTED_VERSION_ERROR_MESSAGE. Could UNRELEASED_INSTANCE_ID just be added
as another label in the fall-through group?
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsMembershipManagerTest.java:
##########
@@ -1542,6 +1546,50 @@ public void testOnHeartbeatSuccessWhenInLeaving() {
verify(memberStateListener,
never()).onMemberEpochUpdated(Optional.of(MEMBER_EPOCH + 1),
membershipManager.memberId());
}
+ @ParameterizedTest
+ @MethodSource("staticMemberLeaveOnCloseOperations")
+ public void testStaticMemberUsesExpectedLeaveEpochOnClose(
+ final CloseOptions.GroupMembershipOperation operation,
Review Comment:
localMetrics is never closed. Metrics implements Closeable. The test could
close it in a try-finally or just inline the construction into the
StreamsMembershipManager constructor call like the other static-member tests in
this class do.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
