@Override
    public Optional<OffsetExpirationCondition> offsetExpirationCondition() {
        if (protocolType.isPresent()) {
            if (isInState(EMPTY)) {
                // No members exist in the group =>
                // - If current state timestamp exists and retention period has passed since group became Empty,
                //   expire all offsets with no pending offset commit;
                // - If there is no current state timestamp (old group metadata schema) and retention period has passed
                //   since the last commit timestamp, expire the offset
                return Optional.of(new OffsetExpirationConditionImpl(
                    offsetAndMetadata -> currentStateTimestamp.orElse(offsetAndMetadata.commitTimestampMs))
                );
            } else if (usesConsumerGroupProtocol() && subscribedTopics.isPresent() && isInState(STABLE)) {
                // Consumers exist in the group and group is Stable =>
                // - If the group is aware of the subscribed topics and retention period has passed since the
                //   last commit timestamp, expire the offset.
                return Optional.of(new OffsetExpirationConditionImpl(offsetAndMetadata -> offsetAndMetadata.commitTimestampMs));
            }
        } else {
            // protocolType is None => standalone (simple) consumer, that uses Kafka for offset storage. Only
            // expire offsets where retention period has passed since their last commit.
            return Optional.of(new OffsetExpirationConditionImpl(offsetAndMetadata -> offsetAndMetadata.commitTimestampMs));
        }
        // If none of the conditions above are met, do not expire any offsets.
        return Optional.empty();
    }
@Test
    public void testOffsetExpirationCondition() {
        long currentTimestamp = 30000L;
        long commitTimestamp = 20000L;
        long offsetsRetentionMs = 10000L;
        OffsetAndMetadata offsetAndMetadata = new OffsetAndMetadata(15000L, OptionalInt.empty(), "", commitTimestamp, OptionalLong.empty());
        MockTime time = new MockTime();
        long currentStateTimestamp = time.milliseconds();
        ClassicGroup group = new ClassicGroup(new LogContext(), "groupId", EMPTY, time, mock(GroupCoordinatorMetricsShard.class));

        // 1. Test no protocol type. Simple consumer case, Base timestamp based off of last commit timestamp.
        Optional<OffsetExpirationCondition> offsetExpirationCondition = group.offsetExpirationCondition();
        assertTrue(offsetExpirationCondition.isPresent());

        OffsetExpirationConditionImpl condition = (OffsetExpirationConditionImpl) offsetExpirationCondition.get();
        assertEquals(commitTimestamp, condition.baseTimestamp().apply(offsetAndMetadata));
        assertTrue(condition.isOffsetExpired(offsetAndMetadata, currentTimestamp, offsetsRetentionMs));

        // 2. Test non-consumer protocol type + Empty state. Base timestamp based off of current state timestamp.
        JoinGroupRequestProtocolCollection protocols = new JoinGroupRequestProtocolCollection();
        protocols.add(new JoinGroupRequestProtocol()
            .setName("range")
            .setMetadata(ConsumerProtocol.serializeSubscription(
                new ConsumerPartitionAssignor.Subscription(Collections.singletonList("topic"))).array()));

        ClassicGroupMember memberWithNonConsumerProtocol = new ClassicGroupMember(
            "memberWithNonConsumerProtocol",
            Optional.empty(),
            clientId,
            clientHost,
            rebalanceTimeoutMs,
            sessionTimeoutMs,
            "My Protocol",
            protocols
        );

        group.add(memberWithNonConsumerProtocol);
        assertEquals("My Protocol", group.protocolType().get());

        offsetExpirationCondition = group.offsetExpirationCondition();
        assertTrue(offsetExpirationCondition.isPresent());

        condition = (OffsetExpirationConditionImpl) offsetExpirationCondition.get();
        assertEquals(currentStateTimestamp, condition.baseTimestamp().apply(offsetAndMetadata));
        assertTrue(condition.isOffsetExpired(offsetAndMetadata, currentStateTimestamp + offsetsRetentionMs, offsetsRetentionMs));

        // 3. Test non-consumer protocol type + non-Empty state. Do not expire any offsets.
        group.transitionTo(PREPARING_REBALANCE);
        offsetExpirationCondition = group.offsetExpirationCondition();
        assertFalse(offsetExpirationCondition.isPresent());

        // 4. Test consumer protocol type + subscribed topics + Stable state. Base timestamp based off of last commit timestamp.
        group.remove("memberWithNonConsumerProtocol");
        ClassicGroupMember memberWithConsumerProtocol = new ClassicGroupMember(
            "memberWithConsumerProtocol",
            Optional.empty(),
            clientId,
            clientHost,
            rebalanceTimeoutMs,
            sessionTimeoutMs,
            "consumer",
            protocols
        );
        group.add(memberWithConsumerProtocol);
        group.initNextGeneration();
        group.transitionTo(STABLE);
        assertTrue(group.subscribedTopics().get().contains("topic"));

        offsetExpirationCondition = group.offsetExpirationCondition();
        assertTrue(offsetExpirationCondition.isPresent());

        condition = (OffsetExpirationConditionImpl) offsetExpirationCondition.get();
        assertEquals(commitTimestamp, condition.baseTimestamp().apply(offsetAndMetadata));
        assertTrue(condition.isOffsetExpired(offsetAndMetadata, currentTimestamp, offsetsRetentionMs));

        // 5. Test consumer protocol type + subscribed topics + non-Stable state. Do not expire any offsets.
        group.transitionTo(PREPARING_REBALANCE);
        offsetExpirationCondition = group.offsetExpirationCondition();
        assertFalse(offsetExpirationCondition.isPresent());
    }