public boolean cleanupExpiredOffsets(String groupId, List<CoordinatorRecord> records) {
        TimelineHashMap<String, TimelineHashMap<Integer, OffsetAndMetadata>> offsetsByTopic =
            offsets.offsetsByGroup.get(groupId);
        if (offsetsByTopic == null) {
            return true;
        }

        // We expect the group to exist.
        Group group = groupMetadataManager.group(groupId);
        Set<String> expiredPartitions = new HashSet<>();
        long currentTimestampMs = time.milliseconds();
        Optional<OffsetExpirationCondition> offsetExpirationCondition = group.offsetExpirationCondition();

        if (!offsetExpirationCondition.isPresent()) {
            return false;
        }

        AtomicBoolean allOffsetsExpired = new AtomicBoolean(true);
        OffsetExpirationCondition condition = offsetExpirationCondition.get();

        offsetsByTopic.forEach((topic, partitions) -> {
            if (!group.isSubscribedToTopic(topic)) {
                partitions.forEach((partition, offsetAndMetadata) -> {
                    // We don't expire the offset yet if there is a pending transactional offset for the partition.
                    if (condition.isOffsetExpired(offsetAndMetadata, currentTimestampMs, config.offsetsRetentionMs()) &&
                        !hasPendingTransactionalOffsets(groupId, topic, partition)) {
                        expiredPartitions.add(appendOffsetCommitTombstone(groupId, topic, partition, records).toString());
                        log.debug("[GroupId {}] Expired offset for partition={}-{}", groupId, topic, partition);
                    } else {
                        allOffsetsExpired.set(false);
                    }
                });
            } else {
                allOffsetsExpired.set(false);
            }
        });
        metrics.record(OFFSET_EXPIRED_SENSOR_NAME, expiredPartitions.size());

        // We don't want to remove the group if there are ongoing transactions.
        return allOffsetsExpired.get() && !openTransactionsByGroup.containsKey(groupId);
    }
@Test
    public void testCleanupExpiredOffsets() {
        GroupMetadataManager groupMetadataManager = mock(GroupMetadataManager.class);
        Group group = mock(Group.class);

        OffsetMetadataManagerTestContext context = new OffsetMetadataManagerTestContext.Builder()
            .withGroupMetadataManager(groupMetadataManager)
            .withOffsetsRetentionMinutes(1)
            .build();

        long commitTimestamp = context.time.milliseconds();

        context.commitOffset("group-id", "firstTopic", 0, 100L, 0, commitTimestamp);
        context.commitOffset("group-id", "secondTopic", 0, 100L, 0, commitTimestamp);
        context.commitOffset("group-id", "secondTopic", 1, 100L, 0, commitTimestamp + 500);

        context.time.sleep(Duration.ofMinutes(1).toMillis());

        // firstTopic-0: group is still subscribed to firstTopic. Do not expire.
        // secondTopic-0: should expire as offset retention has passed.
        // secondTopic-1: has not passed offset retention. Do not expire.
        List<CoordinatorRecord> expectedRecords = Collections.singletonList(
            GroupCoordinatorRecordHelpers.newOffsetCommitTombstoneRecord("group-id", "secondTopic", 0)
        );

        when(groupMetadataManager.group("group-id")).thenReturn(group);
        when(group.offsetExpirationCondition()).thenReturn(Optional.of(
            new OffsetExpirationConditionImpl(offsetAndMetadata -> offsetAndMetadata.commitTimestampMs)));
        when(group.isSubscribedToTopic("firstTopic")).thenReturn(true);
        when(group.isSubscribedToTopic("secondTopic")).thenReturn(false);

        List<CoordinatorRecord> records = new ArrayList<>();
        assertFalse(context.cleanupExpiredOffsets("group-id", records));
        assertEquals(expectedRecords, records);

        // Expire secondTopic-1.
        context.time.sleep(500);
        expectedRecords = Collections.singletonList(
            GroupCoordinatorRecordHelpers.newOffsetCommitTombstoneRecord("group-id", "secondTopic", 1)
        );

        records = new ArrayList<>();
        assertFalse(context.cleanupExpiredOffsets("group-id", records));
        assertEquals(expectedRecords, records);

        // Add 2 more commits, then expire all.
        when(group.isSubscribedToTopic("firstTopic")).thenReturn(false);
        context.commitOffset("group-id", "firstTopic", 1, 100L, 0, commitTimestamp + 500);
        context.commitOffset("group-id", "secondTopic", 0, 101L, 0, commitTimestamp + 500);

        expectedRecords = Arrays.asList(
            GroupCoordinatorRecordHelpers.newOffsetCommitTombstoneRecord("group-id", "firstTopic", 0),
            GroupCoordinatorRecordHelpers.newOffsetCommitTombstoneRecord("group-id", "firstTopic", 1),
            GroupCoordinatorRecordHelpers.newOffsetCommitTombstoneRecord("group-id", "secondTopic", 0)
        );

        records = new ArrayList<>();
        assertTrue(context.cleanupExpiredOffsets("group-id", records));
        assertEquals(expectedRecords, records);
    }