public CoordinatorResult<OffsetDeleteResponseData, CoordinatorRecord> deleteOffsets(
        OffsetDeleteRequestData request
    ) throws ApiException {
        final Group group = validateOffsetDelete(request);
        final List<CoordinatorRecord> records = new ArrayList<>();
        final OffsetDeleteResponseData.OffsetDeleteResponseTopicCollection responseTopicCollection =
            new OffsetDeleteResponseData.OffsetDeleteResponseTopicCollection();

        request.topics().forEach(topic -> {
            final OffsetDeleteResponseData.OffsetDeleteResponsePartitionCollection responsePartitionCollection =
                new OffsetDeleteResponseData.OffsetDeleteResponsePartitionCollection();

            if (group.isSubscribedToTopic(topic.name())) {
                topic.partitions().forEach(partition ->
                    responsePartitionCollection.add(new OffsetDeleteResponseData.OffsetDeleteResponsePartition()
                        .setPartitionIndex(partition.partitionIndex())
                        .setErrorCode(Errors.GROUP_SUBSCRIBED_TO_TOPIC.code())
                    )
                );
            } else {
                topic.partitions().forEach(partition -> {
                    // We always add the partition to the response.
                    responsePartitionCollection.add(new OffsetDeleteResponseData.OffsetDeleteResponsePartition()
                        .setPartitionIndex(partition.partitionIndex())
                    );

                    // A tombstone is written if an offset is present in the main storage or
                    // if a pending transactional offset exists.
                    if (hasCommittedOffset(request.groupId(), topic.name(), partition.partitionIndex()) ||
                        hasPendingTransactionalOffsets(request.groupId(), topic.name(), partition.partitionIndex())) {
                        records.add(CoordinatorRecordHelpers.newOffsetCommitTombstoneRecord(
                            request.groupId(),
                            topic.name(),
                            partition.partitionIndex()
                        ));
                    }
                });
            }

            responseTopicCollection.add(new OffsetDeleteResponseData.OffsetDeleteResponseTopic()
                .setName(topic.name())
                .setPartitions(responsePartitionCollection)
            );
        });
        metrics.record(OFFSET_DELETIONS_SENSOR_NAME, records.size());

        return new CoordinatorResult<>(
            records,
            new OffsetDeleteResponseData().setTopics(responseTopicCollection)
        );
    }
Handles an OffsetDelete request.

@param request The OffsetDelete request.

@return A Result containing the OffsetDeleteResponseData response and
        a list of records to update the state machine.
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetMetadataManager.java