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 testCleanupExpiredOffsetsGroupHasNoOffsets() {
        OffsetMetadataManagerTestContext context = new OffsetMetadataManagerTestContext.Builder()
            .build();

        List<CoordinatorRecord> records = new ArrayList<>();
        assertTrue(context.cleanupExpiredOffsets("unknown-group-id", records));
        assertEquals(Collections.emptyList(), records);
    }