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

[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.