public CoordinatorResult<Void, CoordinatorRecord> cleanupGroupMetadata() {
        long startMs = time.milliseconds();
        List<CoordinatorRecord> records = new ArrayList<>();
        groupMetadataManager.groupIds().forEach(groupId -> {
            boolean allOffsetsExpired = offsetMetadataManager.cleanupExpiredOffsets(groupId, records);
            if (allOffsetsExpired) {
                groupMetadataManager.maybeDeleteGroup(groupId, records);
            }
        });

        log.info("Generated {} tombstone records while cleaning up group metadata in {} milliseconds.",
            records.size(), time.milliseconds() - startMs);
        // Reschedule the next cycle.
        scheduleGroupMetadataExpiration();
        return new CoordinatorResult<>(records, false);
    }
@Test
    public void testCleanupGroupMetadata() {
        GroupMetadataManager groupMetadataManager = mock(GroupMetadataManager.class);
        OffsetMetadataManager offsetMetadataManager = mock(OffsetMetadataManager.class);
        Time mockTime = new MockTime();
        MockCoordinatorTimer<Void, CoordinatorRecord> timer = new MockCoordinatorTimer<>(mockTime);
        GroupCoordinatorShard coordinator = new GroupCoordinatorShard(
            new LogContext(),
            groupMetadataManager,
            offsetMetadataManager,
            mockTime,
            timer,
            mock(GroupCoordinatorConfig.class),
            mock(CoordinatorMetrics.class),
            mock(CoordinatorMetricsShard.class)
        );

        CoordinatorRecord offsetCommitTombstone = GroupCoordinatorRecordHelpers.newOffsetCommitTombstoneRecord("group-id", "topic", 0);
        CoordinatorRecord groupMetadataTombstone = GroupCoordinatorRecordHelpers.newGroupMetadataTombstoneRecord("group-id");

        @SuppressWarnings("unchecked")
        ArgumentCaptor<List<CoordinatorRecord>> recordsCapture = ArgumentCaptor.forClass(List.class);

        when(groupMetadataManager.groupIds()).thenReturn(mkSet("group-id", "other-group-id"));
        when(offsetMetadataManager.cleanupExpiredOffsets(eq("group-id"), recordsCapture.capture()))
            .thenAnswer(invocation -> {
                List<CoordinatorRecord> records = recordsCapture.getValue();
                records.add(offsetCommitTombstone);
                return true;
            });
        when(offsetMetadataManager.cleanupExpiredOffsets("other-group-id", Collections.emptyList())).thenReturn(false);
        doAnswer(invocation -> {
            List<CoordinatorRecord> records = recordsCapture.getValue();
            records.add(groupMetadataTombstone);
            return null;
        }).when(groupMetadataManager).maybeDeleteGroup(eq("group-id"), recordsCapture.capture());

        assertFalse(timer.contains(GROUP_EXPIRATION_KEY));
        CoordinatorResult<Void, CoordinatorRecord> result = coordinator.cleanupGroupMetadata();
        assertTrue(timer.contains(GROUP_EXPIRATION_KEY));

        List<CoordinatorRecord> expectedRecords = Arrays.asList(offsetCommitTombstone, groupMetadataTombstone);
        assertEquals(expectedRecords, result.records());
        assertNull(result.response());
        assertNull(result.appendFuture());

        verify(groupMetadataManager, times(1)).groupIds();
        verify(offsetMetadataManager, times(1)).cleanupExpiredOffsets(eq("group-id"), any());
        verify(offsetMetadataManager, times(1)).cleanupExpiredOffsets(eq("other-group-id"), any());
        verify(groupMetadataManager, times(1)).maybeDeleteGroup(eq("group-id"), any());
        verify(groupMetadataManager, times(0)).maybeDeleteGroup(eq("other-group-id"), any());
    }