@Override
    public boolean isOffsetExpired(OffsetAndMetadata offset, long currentTimestampMs, long offsetsRetentionMs) {
        if (offset.expireTimestampMs.isPresent()) {
            // Older versions with explicit expire_timestamp field => old expiration semantics is used
            return currentTimestampMs >= offset.expireTimestampMs.getAsLong();
        } else {
            // Current version with no per partition retention
            return currentTimestampMs - baseTimestamp.apply(offset) >= offsetsRetentionMs;
        }
    }
@Test
    public void testIsOffsetExpired() {
        long currentTimestamp = 1500L;
        long commitTimestamp = 500L;
        OptionalLong expireTimestampMs = OptionalLong.of(1500);
        long offsetsRetentionMs = 500L;

        OffsetExpirationConditionImpl condition = new OffsetExpirationConditionImpl(__ -> commitTimestamp);
        OffsetAndMetadata offsetAndMetadata = new OffsetAndMetadata(
            100,
            OptionalInt.of(1),
            "metadata",
            commitTimestamp,
            expireTimestampMs
        );

        // Test when expire timestamp exists (older versions with per partition retention)
        // 1. Current timestamp >= expire timestamp => should expire
        assertTrue(condition.isOffsetExpired(offsetAndMetadata, currentTimestamp, offsetsRetentionMs));

        // 2. Current timestamp < expire timestamp => should not expire
        currentTimestamp = 499;
        assertFalse(condition.isOffsetExpired(offsetAndMetadata, currentTimestamp, offsetsRetentionMs));

        // Test when expire timestamp does not exist (current version with no per partition retention)
        offsetAndMetadata = new OffsetAndMetadata(
            100,
            OptionalInt.of(1),
            "metadata",
            commitTimestamp,
            OptionalLong.empty()
        );

        // 3. Current timestamp - base timestamp >= offsets retention => should expire
        currentTimestamp = 1000L;
        assertTrue(condition.isOffsetExpired(offsetAndMetadata, currentTimestamp, offsetsRetentionMs));

        // 4. Current timestamp - base timestamp < offsets retention => should not expire
        currentTimestamp = 999L;
        assertFalse(condition.isOffsetExpired(offsetAndMetadata, currentTimestamp, offsetsRetentionMs));
    }