public List<CoordinatorRecord> onPartitionsDeleted(
        List<TopicPartition> topicPartitions
    ) {
        List<CoordinatorRecord> records = new ArrayList<>();

        Map<String, List<Integer>> partitionsByTopic = new HashMap<>();
        topicPartitions.forEach(tp -> partitionsByTopic
            .computeIfAbsent(tp.topic(), __ -> new ArrayList<>())
            .add(tp.partition())
        );

        Consumer<Offsets> delete = offsetsToClean -> {
            offsetsToClean.offsetsByGroup.forEach((groupId, topicOffsets) -> {
                topicOffsets.forEach((topic, partitionOffsets) -> {
                    if (partitionsByTopic.containsKey(topic)) {
                        partitionsByTopic.get(topic).forEach(partition -> {
                            if (partitionOffsets.containsKey(partition)) {
                                appendOffsetCommitTombstone(groupId, topic, partition, records);
                            }
                        });
                    }
                });
            });
        };

        // Delete the partitions from the main storage.
        delete.accept(offsets);
        // Delete the partitions from the pending transactional offsets.
        pendingTransactionalOffsets.forEach((__, offsets) -> delete.accept(offsets));

        return records;
    }
@Test
    public void testOnPartitionsDeleted() {
        OffsetMetadataManagerTestContext context = new OffsetMetadataManagerTestContext.Builder().build();

        // Commit offsets.
        context.commitOffset("grp-0", "foo", 1, 100, 1, context.time.milliseconds());
        context.commitOffset("grp-0", "foo", 2, 200, 1, context.time.milliseconds());
        context.commitOffset("grp-0", "foo", 3, 300, 1, context.time.milliseconds());

        context.commitOffset("grp-1", "bar", 1, 100, 1, context.time.milliseconds());
        context.commitOffset("grp-1", "bar", 2, 200, 1, context.time.milliseconds());
        context.commitOffset("grp-1", "bar", 3, 300, 1, context.time.milliseconds());

        context.commitOffset(100L, "grp-2", "foo", 1, 100, 1, context.time.milliseconds());
        context.commitOffset(100L, "grp-2", "foo", 2, 200, 1, context.time.milliseconds());
        context.commitOffset(100L, "grp-2", "foo", 3, 300, 1, context.time.milliseconds());

        // Delete partitions.
        List<CoordinatorRecord> records = context.deletePartitions(Arrays.asList(
            new TopicPartition("foo", 1),
            new TopicPartition("foo", 2),
            new TopicPartition("foo", 3),
            new TopicPartition("bar", 1)
        ));

        // Verify.
        List<CoordinatorRecord> expectedRecords = Arrays.asList(
            GroupCoordinatorRecordHelpers.newOffsetCommitTombstoneRecord("grp-0", "foo", 1),
            GroupCoordinatorRecordHelpers.newOffsetCommitTombstoneRecord("grp-0", "foo", 2),
            GroupCoordinatorRecordHelpers.newOffsetCommitTombstoneRecord("grp-0", "foo", 3),
            GroupCoordinatorRecordHelpers.newOffsetCommitTombstoneRecord("grp-1", "bar", 1),
            GroupCoordinatorRecordHelpers.newOffsetCommitTombstoneRecord("grp-2", "foo", 1),
            GroupCoordinatorRecordHelpers.newOffsetCommitTombstoneRecord("grp-2", "foo", 2),
            GroupCoordinatorRecordHelpers.newOffsetCommitTombstoneRecord("grp-2", "foo", 3)
        );

        assertEquals(new HashSet<>(expectedRecords), new HashSet<>(records));

        assertFalse(context.hasOffset("grp-0", "foo", 1));
        assertFalse(context.hasOffset("grp-0", "foo", 2));
        assertFalse(context.hasOffset("grp-0", "foo", 3));
        assertFalse(context.hasOffset("grp-1", "bar", 1));
        assertFalse(context.hasOffset("grp-2", "foo", 1));
        assertFalse(context.hasOffset("grp-2", "foo", 2));
        assertFalse(context.hasOffset("grp-2", "foo", 3));
    }