expect always earliest offset on empty initial offset for the records

plugins/pulsar-records-storage/src/test/java/com/github/bsideup/liiklus/pulsar/PulsarRecordsStorageTest.java

@@ -4,6 +4,7 @@
 import com.github.bsideup.liiklus.records.RecordStorageTests;
 import com.github.bsideup.liiklus.records.RecordsStorage;
 import com.github.bsideup.liiklus.records.RecordsStorage.PartitionSource;
+import com.github.bsideup.liiklus.support.DisabledUntil;
 import lombok.Getter;
 import lombok.SneakyThrows;
 import org.apache.pulsar.client.admin.PulsarAdmin;
@@ -79,14 +80,21 @@ public int getNumberOfPartitions() {
         return NUM_OF_PARTITIONS;
     }
 
+    @Override
+    @Test
+    @DisabledUntil(value = "2020-01-01", comment = "#180 - Pulsar should fix the way seek works, not disconnecting consumers (apache/pulsar/pull/5022)")
+    public void shouldAlwaysUseEarliestOffsetOnEmptyOffsetsInTheInitialProvider() {
+        RecordStorageTests.super.shouldAlwaysUseEarliestOffsetOnEmptyOffsetsInTheInitialProvider();
+    }
+
     @Test
     void shouldPreferEventTimeOverPublishTime() throws Exception {
         var topic = getTopic();
         var partition = 0;
         var key = keyByPartition(partition);
         var eventTimestamp = Instant.now().minusSeconds(1000).truncatedTo(ChronoUnit.MILLIS);
 
-        try(
+        try (
                 var pulsarClient = PulsarClient.builder()
                         .serviceUrl(pulsar.getPulsarBrokerUrl())
                         .build()

tck/src/main/java/com/github/bsideup/liiklus/records/tests/ConsumerGroupTest.java

@@ -1,6 +1,7 @@
 package com.github.bsideup.liiklus.records.tests;
 
 import com.github.bsideup.liiklus.records.RecordStorageTestSupport;
+import com.github.bsideup.liiklus.records.RecordsStorage;
 import com.github.bsideup.liiklus.records.RecordsStorage.OffsetInfo;
 import com.github.bsideup.liiklus.records.RecordsStorage.PartitionSource;
 import com.github.bsideup.liiklus.records.RecordsStorage.Subscription;
@@ -12,16 +13,20 @@
 import reactor.util.function.Tuple2;
 import reactor.util.function.Tuples;
 
+import java.time.Duration;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Function;
+import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
@@ -170,4 +175,43 @@ default void testExclusiveRecordDistribution() throws Exception {
             disposeAll.onNext(true);
         }
     }
+
+
+    @Test
+    default void shouldAlwaysUseEarliestOffsetOnEmptyOffsetsInTheInitialProvider() {
+        String groupName = "name";
+        var earliest = "earliest";
+        int count = 10;
+        int offsetShift = 5;
+
+        var published = publishMany(UUID.randomUUID().toString().getBytes(), count);
+        var latest = published.get(published.size() - 1);
+
+        RecordsStorage.Record latestRecord = subscribeToPartitionWithGroup(groupName, latest.getPartition(), earliest, () -> CompletableFuture.completedStage(Map.of(latest.getPartition(), latest.getOffset() - offsetShift)))
+                .flatMap(PartitionSource::getPublisher)
+                .takeUntil(next -> next.getOffset() == latest.getOffset())
+                .blockLast(Duration.ofSeconds(10));
+
+        assertThat(latestRecord.getOffset()).describedAs("latest offset").isEqualTo(latest.getOffset());
+
+        List<RecordsStorage.Record> records = subscribeToPartitionWithGroup(groupName, latest.getPartition(), earliest, () -> CompletableFuture.completedStage(Map.of()))
+                .flatMap(PartitionSource::getPublisher)
+                .takeUntil(next -> next.getOffset() == latest.getOffset())
+                .collectList()
+                .block(Duration.ofSeconds(10));
+
+        assertThat(records).hasSize(count);
+    }
+
+
+    default Flux<? extends RecordsStorage.PartitionSource> subscribeToPartitionWithGroup(
+            String groupName,
+            int partition,
+            String offsetReset,
+            Supplier<CompletionStage<Map<Integer, Long>>> offsetsProvider
+    ) {
+        return Flux.from(getTarget().subscribe(getTopic(), groupName, Optional.of(offsetReset)).getPublisher(offsetsProvider))
+                .flatMapIterable(it -> it::iterator)
+                .filter(it -> partition == it.getPartition());
+    }
 }

tck/src/main/java/com/github/bsideup/liiklus/support/DisabledUntil.java

@@ -0,0 +1,42 @@
+package com.github.bsideup.liiklus.support;
+
+import org.junit.jupiter.api.extension.ConditionEvaluationResult;
+import org.junit.jupiter.api.extension.ExecutionCondition;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.ExtensionContext;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+import java.time.LocalDate;
+import java.util.Optional;
+
+import static org.junit.platform.commons.support.AnnotationSupport.findAnnotation;
+
+
+@Target({ElementType.TYPE, ElementType.METHOD})
+@Retention(RetentionPolicy.RUNTIME)
+@ExtendWith(DisabledUntil.DisabledCondition.class)
+public @interface DisabledUntil {
+    /**
+     * Local date, ISO formatted string such as 2000-01-30
+     */
+    String value();
+
+    String comment();
+
+    class DisabledCondition implements ExecutionCondition {
+
+        @Override
+        public ConditionEvaluationResult evaluateExecutionCondition(ExtensionContext context) {
+            Optional<DisabledUntil> until = findAnnotation(context.getElement(), DisabledUntil.class);
+            if (until.map(it -> LocalDate.parse(it.value()).isAfter(LocalDate.now())).orElse(false)) {
+                String reason = until.map(DisabledUntil::comment).orElse("Disabled for now");
+                return ConditionEvaluationResult.disabled(reason);
+            }
+
+            return ConditionEvaluationResult.enabled("Enabled");
+        }
+    }
+}