public int deleteAllOffsets(
        String groupId,
        List<CoordinatorRecord> records
    ) {
        TimelineHashMap<String, TimelineHashMap<Integer, OffsetAndMetadata>> offsetsByTopic = offsets.offsetsByGroup.get(groupId);
        AtomicInteger numDeletedOffsets = new AtomicInteger();

        // Delete all the offsets from the main storage.
        if (offsetsByTopic != null) {
            offsetsByTopic.forEach((topic, offsetsByPartition) ->
                offsetsByPartition.keySet().forEach(partition -> {
                    records.add(CoordinatorRecordHelpers.newOffsetCommitTombstoneRecord(groupId, topic, partition));
                    numDeletedOffsets.getAndIncrement();
                })
            );
        }

        // Delete all the pending transactional offsets too. Here we only write a tombstone
        // if the topic-partition was not in the main storage because we don't need to write
        // two consecutive tombstones.
        TimelineHashSet<Long> openTransactions = openTransactionsByGroup.get(groupId);
        if (openTransactions != null) {
            openTransactions.forEach(producerId -> {
                Offsets pendingOffsets = pendingTransactionalOffsets.get(producerId);
                if (pendingOffsets != null) {
                    TimelineHashMap<String, TimelineHashMap<Integer, OffsetAndMetadata>> pendingGroupOffsets =
                        pendingOffsets.offsetsByGroup.get(groupId);
                    if (pendingGroupOffsets != null) {
                        pendingGroupOffsets.forEach((topic, offsetsByPartition) -> {
                            offsetsByPartition.keySet().forEach(partition -> {
                                if (!hasCommittedOffset(groupId, topic, partition)) {
                                    records.add(CoordinatorRecordHelpers.newOffsetCommitTombstoneRecord(groupId, topic, partition));
                                    numDeletedOffsets.getAndIncrement();
                                }
                            });
                        });
                    }
                }
            });
        }

        return numDeletedOffsets.get();
    }
Deletes offsets as part of a DeleteGroups request.
Populates the record list passed in with records to update the state machine.
Validations are done in {@link GroupCoordinatorShard#deleteGroups(RequestContext, List)}

@param groupId The id of the given group.
@param records The record list to populate.

@return The number of offsets to be deleted.
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetMetadataManager.java