Repository: apache/kafka
File: group-coordinator/src/test/java/org/apache/kafka/coordinator/group/OffsetMetadataManagerTest.java (Java)

Code before:
        context.testOffsetDeleteWith("foo", "bar", 0, Errors.GROUP_SUBSCRIBED_TO_TOPIC);
    }

    @Test
    public void testConsumerGroupOffsetDeleteWithPendingTransactionalOffsets() {
        OffsetMetadataManagerTestContext context = new OffsetMetadataManagerTestContext.Builder().build();
        ConsumerGroup group = context.groupMetadataManager.getOrMaybeCreatePersistedConsumerGroup(
            "foo",
            true
        );
        verifyOffsetDeleteWithPendingTransactionalOffsets(context, group);
    }
    
    @Test
    public void testStreamsGroupOffsetDeleteWithPendingTransactionalOffsets() {
        OffsetMetadataManagerTestContext context = new OffsetMetadataManagerTestContext.Builder().build();
        StreamsGroup group = context.groupMetadataManager.getOrMaybeCreatePersistedStreamsGroup(
            "foo",
            true
        );
        verifyOffsetDeleteWithPendingTransactionalOffsets(context, group);
    }

    private static void verifyOffsetDeleteWithPendingTransactionalOffsets(OffsetMetadataManagerTestContext context, Group group) {
        context.commitOffset(10L, "foo", "bar", 0, 100L, 0, context.time.milliseconds());
        assertFalse(group.isSubscribedToTopic("bar"));
        context.testOffsetDeleteWith("foo", "bar", 0, Errors.NONE);
        assertFalse(context.hasOffset("foo", "bar", 0));
    }

    @ParameterizedTest
    @EnumSource(value = Group.GroupType.class, names = {"CLASSIC", "CONSUMER", "STREAMS"})
    public void testDeleteGroupAllOffsets(Group.GroupType groupType) {
        OffsetMetadataManagerTestContext context = new OffsetMetadataManagerTestContext.Builder().build();
        context.getOrMaybeCreateGroup(groupType, "foo");

        context.commitOffset("foo", "bar-0", 0, 100L, 0);
        context.commitOffset("foo", "bar-0", 1, 100L, 0);
        context.commitOffset("foo", "bar-1", 0, 100L, 0);

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

        List<CoordinatorRecord> records = new ArrayList<>();
        int numDeleteOffsets = context.deleteAllOffsets("foo", records);

        assertEquals(expectedRecords, records);
        assertEquals(3, numDeleteOffsets);

Review comment (suggestion):
Some existing tests used this kind of parametrization. I used it as well in the test where we already had it, however, in the general case the setup was often a little different between the group types or existing parametrization made adopting this pattern painful, so I opted for a separate test method above but shared everything after the setup.

Code after:
        ConsumerGroup group = context.groupMetadataManager.getOrMaybeCreatePersistedConsumerGroup(
            "foo",
            true
        );
        verifyOffsetDeleteWithPendingTransactionalOffsets(context, group);
    }
    
    @Test
    public void testStreamsGroupOffsetDeleteWithPendingTransactionalOffsets() {
        OffsetMetadataManagerTestContext context = new OffsetMetadataManagerTestContext.Builder().build();
        StreamsGroup group = context.groupMetadataManager.getOrMaybeCreatePersistedStreamsGroup(
            "foo",
            true
        );
        verifyOffsetDeleteWithPendingTransactionalOffsets(context, group);
    }

    private static void verifyOffsetDeleteWithPendingTransactionalOffsets(OffsetMetadataManagerTestContext context, Group group) {
        context.commitOffset(10L, "foo", "bar", 0, 100L, 0, context.time.milliseconds());
        assertFalse(group.isSubscribedToTopic("bar"));
        context.testOffsetDeleteWith("foo", "bar", 0, Errors.NONE);
        assertFalse(context.hasOffset("foo", "bar", 0));
    }

    @ParameterizedTest
    @EnumSource(value = Group.GroupType.class, names = {"CLASSIC", "CONSUMER", "STREAMS"})
    public void testDeleteGroupAllOffsets(Group.GroupType groupType) {
        OffsetMetadataManagerTestContext context = new OffsetMetadataManagerTestContext.Builder().build();
        context.getOrMaybeCreateGroup(groupType, "foo");

        context.commitOffset("foo", "bar-0", 0, 100L, 0);
        context.commitOffset("foo", "bar-0", 1, 100L, 0);
        context.commitOffset("foo", "bar-1", 0, 100L, 0);

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

        List<CoordinatorRecord> records = new ArrayList<>();
        int numDeleteOffsets = context.deleteAllOffsets("foo", records);

        assertEquals(expectedRecords, records);
        assertEquals(3, numDeleteOffsets);
    }

    @ParameterizedTest
    @EnumSource(value = Group.GroupType.class, names = {"CLASSIC", "CONSUMER", "STREAMS"})
    public void testDeleteGroupAllOffsetsWithPendingTransactionalOffsets(Group.GroupType groupType) {
        OffsetMetadataManagerTestContext context = new OffsetMetadataManagerTestContext.Builder().build();