KAFKA-19440: Handle top-level errors in AlterShareGroupOffsets RPC

[AndrewJSchofield]

While testing the code in #19820, it
became clear that the error handling problems were due to the underlying
Admin API. This PR fixes the error handling for top-level errors in the
AlterShareGroupOffsets RPC.

Reviewers: Apoorv Mittal apoorvmittal10@gmail.com, Lan Ding
isDing_L@163.com, TaiJuWu tjwu1217@gmail.com

[DL1231]

Thanks for the patch, just out of curiosity, would the AlterConsumerGroupOffsets RPC have the same issue?

kafka/clients/src/main/java/org/apache/kafka/clients/admin/AlterConsumerGroupOffsetsResult.java

Lines 68 to 79 in 05b6e81

	return this.future.thenApply(topicPartitionErrorsMap -> {
	List<TopicPartition> partitionsFailed = topicPartitionErrorsMap.entrySet()
	.stream()
	.filter(e -> e.getValue() != Errors.NONE)
	.map(Map.Entry::getKey)
	.collect(Collectors.toList());
	for (Errors error : topicPartitionErrorsMap.values()) {
	if (error != Errors.NONE) {
	throw error.exception(
	"Failed altering group offsets for the following partitions: " + partitionsFailed);
	}
	}

[AndrewJSchofield]

I'll check it out before I merge, but the important difference here is in KafkaApis.scala. For DeleteSGO, it already handled a non-zero error code. For AlterSGO, that code was missing.

[AndrewJSchofield]

Ah, sorry. I didn't read your comment accurately. The difference is that AlterShareGroupOffsets can successfully pass back an error code, which is why this is in terms of ApiException rather than Errors. For AlterConsumerGroupOffsets, the RPC is actually OffsetCommit and this does not have an ErrorMessage field at all. So, it cannot be fixed for consumer groups until we have a version bump on the OffsetCommit RPC.

[TaiJuWu]

Should this change to immutable?

[AndrewJSchofield]

I'm not sure it matters here. The list in internal to this method, and it is just converted to a string and appended to the exception message.

[DL1231]

@AndrewJSchofield The last build failed with AuthorizerIntegrationTest failed. Could you please check?

[AndrewJSchofield]

@AndrewJSchofield The last build failed with AuthorizerIntegrationTest failed. Could you please check?

Interesting. I fixed a mistake and I expect the test needs fixing too. Will do.

[DL1231]

Thanks for the patch, LGTM.

[TaiJuWu]

LGTM.

[apoorvmittal10]

Thanks for the PR, LGTM!

[chia7712]

nit: we can reuse the Errors

[chia7712]

nit: the debug message should use placeholder for partitionErrorMessage

[chia7712]

for example:

                var partitionError = Errors.forCode(partition.errorCode());
                var partitionErrorMessage = partition.errorMessage();
                if (partitionError != Errors.NONE) {
                    log.debug("AlterShareGroupOffsets request for group id {} and topic-partition {}-{} failed and returned error {}. {}",
                        groupId.idValue, topic.topicName(), partition.partitionIndex(), partitionError, partitionErrorMessage);
                }
                partitionResults.put(new TopicPartition(topic.topicName(), partition.partitionIndex()), partitionError.exception(partitionErrorMessage));

[AndrewJSchofield]

Thanks @chia7712 . I'll put in a minor PR today.