smjn commented on code in PR #19478:
URL: https://github.com/apache/kafka/pull/19478#discussion_r2046219686
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java:
##########
@@ -704,30 +710,102 @@ public DeleteShareGroupOffsetsResultHolder
shareGroupDeleteOffsetsRequest(
groupMetadataManager.sharePartitionsEligibleForOffsetDeletion(
groupId,
requestData,
- errorTopicResponseList
+ errorTopicResponseList,
+ records
);
if (deleteShareGroupStateRequestTopicsData.isEmpty()) {
- return new
DeleteShareGroupOffsetsResultHolder(Errors.NONE.code(), null,
errorTopicResponseList);
+ return new CoordinatorResult<>(
+ records,
+ new
DeleteShareGroupOffsetsResultHolder(Errors.NONE.code(), null,
errorTopicResponseList)
+ );
}
DeleteShareGroupStateRequestData deleteShareGroupStateRequestData
= new DeleteShareGroupStateRequestData()
.setGroupId(requestData.groupId())
.setTopics(deleteShareGroupStateRequestTopicsData);
- return new DeleteShareGroupOffsetsResultHolder(
- Errors.NONE.code(),
- null,
- errorTopicResponseList,
-
DeleteShareGroupStateParameters.from(deleteShareGroupStateRequestData)
+ return new CoordinatorResult<>(
+ records,
+ new DeleteShareGroupOffsetsResultHolder(
+ Errors.NONE.code(),
+ null,
+ errorTopicResponseList,
+
DeleteShareGroupStateParameters.from(deleteShareGroupStateRequestData)
+ )
);
} catch (GroupIdNotFoundException exception) {
log.error("groupId {} not found", groupId, exception);
- return new
DeleteShareGroupOffsetsResultHolder(Errors.GROUP_ID_NOT_FOUND.code(),
exception.getMessage());
+ return new CoordinatorResult<>(
+ records,
+ new
DeleteShareGroupOffsetsResultHolder(Errors.GROUP_ID_NOT_FOUND.code(),
exception.getMessage())
+ );
} catch (GroupNotEmptyException exception) {
log.error("Provided group {} is not empty", groupId);
- return new
DeleteShareGroupOffsetsResultHolder(Errors.NON_EMPTY_GROUP.code(),
exception.getMessage());
+ return new CoordinatorResult<>(
+ records,
+ new
DeleteShareGroupOffsetsResultHolder(Errors.NON_EMPTY_GROUP.code(),
exception.getMessage())
+ );
+ }
+ }
+
+ /**
+ * Completes the share group offset deletion by creating a
ShareGroupStatePartitionMetadataRecord removing the
+ * deleted topics from deletingTopics set. Returns the final response for
DeleteShareGroupOffsetsRequest
+ *
+ * @param groupId - The group ID
+ * @param topics - The set of topics which were deleted successfully by
the persister
+ * @return the final response {@link DeleteShareGroupOffsetsResponseData}
for the DeleteShareGroupOffsetsRequest
+ */
+ public CoordinatorResult<DeleteShareGroupOffsetsResponseData,
CoordinatorRecord> completeDeleteShareGroupOffsets(
+ String groupId,
+ Map<Uuid, String> topics,
+
List<DeleteShareGroupOffsetsResponseData.DeleteShareGroupOffsetsResponseTopic>
errorTopicResponseList
+ ) {
+ List<CoordinatorRecord> records = new ArrayList<>();
+ try {
+ ShareGroup group = groupMetadataManager.shareGroup(groupId);
+ group.validateDeleteGroup();
+
+
List<DeleteShareGroupOffsetsResponseData.DeleteShareGroupOffsetsResponseTopic>
topicResponseList = new ArrayList<>(List.of());
Review Comment:
why initialize with List.of()?
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
