add GetEndOffsets API

api/src/main/java/com/github/bsideup/liiklus/records/FiniteRecordsStorage.java

@@ -0,0 +1,13 @@
+package com.github.bsideup.liiklus.records;
+
+import java.util.Map;
+import java.util.concurrent.CompletionStage;
+
+public interface FiniteRecordsStorage extends RecordsStorage {
+
+    /**
+     * Returns a {@link Map} where key is partition's number and value is the latest offset.
+     * The offset can be zero. Offset -1 means that there is no offset for this partition.
+     */
+    CompletionStage<Map<Integer, Long>> getEndOffsets(String topic);
+}

app/src/main/java/com/github/bsideup/liiklus/service/ReactorLiiklusServiceImpl.java

@@ -5,6 +5,7 @@
 import com.github.bsideup.liiklus.positions.GroupId;
 import com.github.bsideup.liiklus.positions.PositionsStorage;
 import com.github.bsideup.liiklus.protocol.*;
+import com.github.bsideup.liiklus.records.FiniteRecordsStorage;
 import com.github.bsideup.liiklus.records.RecordPostProcessor;
 import com.github.bsideup.liiklus.records.RecordPreProcessor;
 import com.github.bsideup.liiklus.records.RecordsStorage;
@@ -302,6 +303,24 @@ public Mono<GetOffsetsReply> getOffsets(Mono<GetOffsetsRequest> request) {
         );
     }
 
+    @Override
+    public Mono<GetEndOffsetsReply> getEndOffsets(GetEndOffsetsRequest message, ByteBuf metadata) {
+        return getEndOffsets(Mono.just(message));
+    }
+
+    @Override
+    public Mono<GetEndOffsetsReply> getEndOffsets(Mono<GetEndOffsetsRequest> request) {
+        return request.flatMap(getEndOffsets -> {
+            if (!(recordsStorage instanceof FiniteRecordsStorage)) {
+                return Mono.error(Status.INTERNAL.withDescription("The record storage is not finite").asException());
+            }
+
+            var topic = getEndOffsets.getTopic();
+            return Mono.fromCompletionStage(((FiniteRecordsStorage) recordsStorage).getEndOffsets(topic))
+                    .map(endOffsets -> GetEndOffsetsReply.newBuilder().putAllOffsets(endOffsets).build());
+        });
+    }
+
     private Mono<Map<Integer, Optional<Long>>> getLatestOffsetsOfGroup(String topic, String groupName) {
         return getOffsetsByGroupName(topic, groupName)
                 .map(ackedOffsets -> ackedOffsets.values().stream()

app/src/test/java/com/github/bsideup/liiklus/EndOffsetsTest.java

@@ -0,0 +1,57 @@
+package com.github.bsideup.liiklus;
+
+import com.github.bsideup.liiklus.protocol.GetEndOffsetsRequest;
+import com.github.bsideup.liiklus.protocol.PublishRequest;
+import com.github.bsideup.liiklus.test.AbstractIntegrationTest;
+import com.google.protobuf.ByteString;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.UUID;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class EndOffsetsTest extends AbstractIntegrationTest {
+
+    private String topic;
+
+    @Before
+    public final void setUpEndOffsetsTest() {
+        topic = testName.getMethodName();
+    }
+
+    @Test
+    public void testEndOffsets() {
+        var value = ByteString.copyFromUtf8("foo");
+
+        for (int partition = 0; partition < NUM_PARTITIONS; partition++) {
+            for (int i = 0; i < partition + 1; i++) {
+                stub.publish(PublishRequest.newBuilder()
+                        .setTopic(topic)
+                        .setKey(ByteString.copyFromUtf8(PARTITION_KEYS.get(partition)))
+                        .setValue(value)
+                        .build()
+                ).block();
+            }
+        }
+
+        var reply = stub.getEndOffsets(GetEndOffsetsRequest.newBuilder().setTopic(topic).build()).block();
+
+        assertThat(reply.getOffsetsMap())
+                .hasSize(NUM_PARTITIONS)
+                .allSatisfy((partition, offset) -> {
+                    assertThat(offset)
+                            .as("offset of p" + partition)
+                            .isEqualTo(partition.longValue());
+                });
+    }
+
+    @Test
+    public void testEndOffsets_unknownTopic() {
+        var randomTopic = UUID.randomUUID().toString();
+        var reply = stub.getEndOffsets(GetEndOffsetsRequest.newBuilder().setTopic(randomTopic).build()).block();
+
+        assertThat(reply.getOffsetsMap()).isEmpty();
+    }
+
+}

client/src/main/java/com/github/bsideup/liiklus/LiiklusClient.java

@@ -16,4 +16,6 @@ public interface LiiklusClient {
     Mono<Empty> ack(AckRequest message);
 
     Mono<GetOffsetsReply> getOffsets(GetOffsetsRequest message);
+
+    Mono<GetEndOffsetsReply> getEndOffsets(GetEndOffsetsRequest message);
 }

client/src/main/java/com/github/bsideup/liiklus/RSocketLiiklusClient.java

@@ -43,4 +43,9 @@ public Mono<Empty> ack(AckRequest message) {
     public Mono<GetOffsetsReply> getOffsets(GetOffsetsRequest message) {
         return liiklusServiceClient.getOffsets(message);
     }
+
+    @Override
+    public Mono<GetEndOffsetsReply> getEndOffsets(GetEndOffsetsRequest message) {
+        return liiklusServiceClient.getEndOffsets(message);
+    }
 }

plugins/inmemory-records-storage/src/main/java/com/github/bsideup/liiklus/records/inmemory/InMemoryRecordsStorage.java

@@ -1,6 +1,6 @@
 package com.github.bsideup.liiklus.records.inmemory;
 
-import com.github.bsideup.liiklus.records.RecordsStorage;
+import com.github.bsideup.liiklus.records.FiniteRecordsStorage;
 import lombok.AccessLevel;
 import lombok.RequiredArgsConstructor;
 import lombok.Value;
@@ -33,7 +33,7 @@
  */
 @RequiredArgsConstructor
 @FieldDefaults(makeFinal = true, level = AccessLevel.PRIVATE)
-public class InMemoryRecordsStorage implements RecordsStorage {
+public class InMemoryRecordsStorage implements FiniteRecordsStorage {
 
     public static int partitionByKey(String key, int numberOfPartitions) {
         return partitionByKey(ByteBuffer.wrap(key.getBytes()), numberOfPartitions);
@@ -74,6 +74,20 @@ public CompletionStage<OffsetInfo> publish(Envelope envelope) {
         ));
     }
 
+    @Override
+    public CompletionStage<Map<Integer, Long>> getEndOffsets(String topic) {
+        var partitions = state.getOrDefault(topic, new StoredTopic(numberOfPartitions)).getPartitions();
+        return CompletableFuture.completedFuture(
+                partitions.entrySet().stream().collect(Collectors.toMap(
+                        Map.Entry::getKey,
+                        it -> Math.max(
+                                0,
+                                it.getValue().getNextOffset().get() - 1
+                        )
+                ))
+        );
+    }
+
     @Override
     public Subscription subscribe(String topic, String groupName, Optional<String> autoOffsetReset) {
         var storedTopic = state.computeIfAbsent(topic, __ -> new StoredTopic(numberOfPartitions));

plugins/kafka-records-storage/src/main/java/com/github/bsideup/liiklus/kafka/KafkaRecordsStorage.java

@@ -1,6 +1,6 @@
 package com.github.bsideup.liiklus.kafka;
 
-import com.github.bsideup.liiklus.records.RecordsStorage;
+import com.github.bsideup.liiklus.records.FiniteRecordsStorage;
 import lombok.SneakyThrows;
 import lombok.Value;
 import lombok.experimental.FieldDefaults;
@@ -33,11 +33,12 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Supplier;
+import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
 @FieldDefaults(makeFinal = true)
 @Slf4j
-public class KafkaRecordsStorage implements RecordsStorage {
+public class KafkaRecordsStorage implements FiniteRecordsStorage {
 
     private static final Scheduler KAFKA_POLL_SCHEDULER = Schedulers.elastic();
 
@@ -60,6 +61,39 @@ public KafkaRecordsStorage(String bootstrapServers) {
         );
     }
 
+    @Override
+    public CompletionStage<Map<Integer, Long>> getEndOffsets(String topic) {
+        return Mono.fromCallable(() -> {
+            var properties = new HashMap<String, Object>();
+            properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+            properties.put(ConsumerConfig.GROUP_ID_CONFIG, UUID.randomUUID().toString());
+            properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
+            properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "0");
+            properties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "1");
+
+            try (
+                    var consumer = new KafkaConsumer<ByteBuffer, ByteBuffer>(
+                            properties,
+                            new ByteBufferDeserializer(),
+                            new ByteBufferDeserializer()
+                    )
+            ) {
+                consumer.subscribe(List.of(topic));
+
+                var endOffsets = consumer.endOffsets(
+                        consumer.partitionsFor(topic).stream()
+                                .map(it -> new TopicPartition(topic, it.partition()))
+                                .collect(Collectors.toSet())
+                );
+
+                return endOffsets.entrySet().stream().collect(Collectors.toMap(
+                        it -> it.getKey().partition(),
+                        it -> it.getValue() - 1
+                ));
+            }
+        }).subscribeOn(Schedulers.elastic()).toFuture();
+    }
+
     @Override
     public CompletionStage<OffsetInfo> publish(Envelope envelope) {
         String topic = envelope.getTopic();

protocol/src/main/proto/LiiklusService.proto

@@ -28,6 +28,10 @@ service LiiklusService {
     rpc GetOffsets(GetOffsetsRequest) returns (GetOffsetsReply) {
 
     }
+
+    rpc GetEndOffsets(GetEndOffsetsRequest) returns (GetEndOffsetsReply) {
+
+    }
 }
 
 message PublishRequest {
@@ -122,4 +126,12 @@ message GetOffsetsRequest {
 
 message GetOffsetsReply {
     map<uint32, uint64> offsets = 1;
+}
+
+message GetEndOffsetsRequest {
+    string topic = 1;
+}
+
+message GetEndOffsetsReply {
+    map<uint32, uint64> offsets = 1;
 }

tck/src/main/java/com/github/bsideup/liiklus/records/RecordStorageTests.java

@@ -2,8 +2,9 @@
 
 import com.github.bsideup.liiklus.records.tests.BackPressureTest;
 import com.github.bsideup.liiklus.records.tests.ConsumerGroupTest;
+import com.github.bsideup.liiklus.records.tests.EndOffsetsTest;
 import com.github.bsideup.liiklus.records.tests.PublishTest;
 import com.github.bsideup.liiklus.records.tests.SubscribeTest;
 
-public interface RecordStorageTests extends PublishTest, SubscribeTest, ConsumerGroupTest, BackPressureTest {
+public interface RecordStorageTests extends PublishTest, SubscribeTest, ConsumerGroupTest, BackPressureTest, EndOffsetsTest {
 }

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

@@ -0,0 +1,67 @@
+package com.github.bsideup.liiklus.records.tests;
+
+import com.github.bsideup.liiklus.records.FiniteRecordsStorage;
+import com.github.bsideup.liiklus.records.RecordStorageTestSupport;
+import org.junit.jupiter.api.Assumptions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInfo;
+
+import java.lang.reflect.Method;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public interface EndOffsetsTest extends RecordStorageTestSupport {
+
+    default FiniteRecordsStorage getFiniteTarget() {
+        return (FiniteRecordsStorage) getTarget();
+    }
+
+    int getNumberOfPartitions();
+
+    String keyByPartition(int partition);
+
+    @BeforeEach
+    default void blah(TestInfo testInfo) {
+        if (EndOffsetsTest.class == testInfo.getTestMethod().map(Method::getDeclaringClass).orElse(null)) {
+            Assumptions.assumeTrue(getTarget() instanceof FiniteRecordsStorage, "target is finite");
+        }
+    }
+
+    @Test
+    default void testEndOffsets() throws Exception {
+        var topic = getTopic();
+
+        for (int partition = 0; partition < getNumberOfPartitions(); partition++) {
+            for (int i = 0; i < partition + 1; i++) {
+                publish(keyByPartition(partition).getBytes(), new byte[1]);
+            }
+        }
+
+        var offsets = getFiniteTarget().getEndOffsets(topic).toCompletableFuture().get(10, TimeUnit.SECONDS);
+
+        assertThat(offsets)
+                .hasSize(getNumberOfPartitions())
+                .allSatisfy((partition, offset) -> {
+                    assertThat(offset)
+                            .as("offset of p" + partition)
+                            .isEqualTo(partition.longValue());
+                });
+    }
+
+    @Test
+    default void testEndOffsets_unknownTopic() throws Exception {
+        var topic = UUID.randomUUID().toString();
+
+        var offsets = getFiniteTarget().getEndOffsets(topic).toCompletableFuture().get(10, TimeUnit.SECONDS);
+
+        assertThat(offsets)
+                .allSatisfy((partition, offset) -> {
+                    assertThat(offset)
+                            .as("offset of p" + partition)
+                            .isEqualTo(-1L);
+                });
+    }
+}