smjn commented on code in PR #19478:
URL: https://github.com/apache/kafka/pull/19478#discussion_r2046223408
##########
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 {
Review Comment:
Is it necessary to add these checks here since this code will be called
after adding deleting?
```
add_deleting (check group exp)
persister_delete
remove_deleting (?)
```
--
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
