[fix] Fix read unstable messages (#1856)

Currently, the consumer can read unstable messages in `read_committed`
mode.

1. If the fetch offset is equal to or greater than LSO(last stable
offset), skip reading entries to avoid useless reading.
2. If the entires' offset is equal to or greater than LSO, discard them
to avoid sending them to the consumer.

Add a new unit to verify consumers can't receive unstable messages.

(cherry picked from commit 0d2ed74)

Author: gaoran10

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/storage/PartitionLog.java

@@ -348,6 +348,19 @@ public static ReadRecordsResult get(DecodeResult decodeResult,
             return readRecordsResult;
         }
 
+        public static ReadRecordsResult empty(long highWatermark,
+                                            long lastStableOffset,
+                                            Position lastPosition,
+                                            PartitionLog partitionLog) {
+            return ReadRecordsResult.get(
+                    DecodeResult.get(MemoryRecords.EMPTY),
+                    Collections.emptyList(),
+                    highWatermark,
+                    lastStableOffset,
+                    lastPosition,
+                    partitionLog);
+        }
+
         public static ReadRecordsResult error(Errors errors, PartitionLog partitionLog) {
             return ReadRecordsResult.error(PositionImpl.EARLIEST, errors, partitionLog);
         }
@@ -587,6 +600,21 @@ public CompletableFuture<ReadRecordsResult> readRecords(final FetchRequestData.F
                 requestStats.getPrepareMetadataStats().registerSuccessfulEvent(
                         MathUtils.elapsedNanos(startPrepareMetadataNanos), TimeUnit.NANOSECONDS);
                 long adjustedMaxBytes = Math.min(partitionData.partitionMaxBytes(), limitBytes.get());
+                if (readCommitted) {
+                    long firstUndecidedOffset = producerStateManager.firstUndecidedOffset().orElse(-1L);
+                    if (firstUndecidedOffset >= 0 && firstUndecidedOffset <= offset) {
+                        long highWaterMark = MessageMetadataUtils.getHighWatermark(cursor.getManagedLedger());
+                        future.complete(
+                                ReadRecordsResult.empty(
+                                        highWaterMark,
+                                        firstUndecidedOffset,
+                                        tcm.getManagedLedger().getLastConfirmedEntry(),
+                                        this
+                                )
+                        );
+                        return;
+                    }
+                }
                 readEntries(cursor, topicPartition, cursorOffset, maxReadEntriesNum, adjustedMaxBytes,
                         fullPartitionName -> {
                     topicManager.invalidateCacheForFencedManagerLedgerOnTopic(fullPartitionName);
@@ -743,7 +771,7 @@ private List<Entry> getCommittedEntries(List<Entry> entries, long lso) {
         committedEntries = new ArrayList<>();
         for (Entry entry : entries) {
             try {
-                if (lso >= MessageMetadataUtils.peekBaseOffsetFromEntry(entry)) {
+                if (lso > MessageMetadataUtils.peekBaseOffsetFromEntry(entry)) {
                     committedEntries.add(entry);
                 } else {
                     break;

tests/src/test/java/io/streamnative/pulsar/handlers/kop/TransactionTest.java

@@ -53,6 +53,7 @@
 import java.util.function.Function;
 import lombok.Cleanup;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.RandomStringUtils;
 import org.apache.kafka.clients.admin.AbortTransactionSpec;
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.admin.DescribeProducersResult;
@@ -1206,6 +1207,7 @@ private KafkaConsumer<Integer, String> buildTransactionConsumer(String groupId,
         consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
         consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
         consumerProps.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, isolation);
+        consumerProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 10);
         addCustomizeProps(consumerProps);
 
         return new KafkaConsumer<>(consumerProps);
@@ -1746,4 +1748,64 @@ protected void addCustomizeProps(Properties producerProps) {
         // No-op
     }
 
+    @DataProvider(name = "isolationProvider")
+    protected Object[][] isolationProvider() {
+        return new Object[][]{
+                {"read_committed"},
+                {"read_uncommitted"},
+        };
+    }
+
+    @Test(dataProvider = "isolationProvider", timeOut = 1000 * 30)
+    public void readUnstableMessagesTest(String isolation) throws InterruptedException, ExecutionException {
+        String topic = "unstable-message-test-" + RandomStringUtils.randomAlphabetic(5);
+
+        KafkaConsumer<Integer, String> consumer = buildTransactionConsumer("unstable-read", isolation);
+        consumer.subscribe(Collections.singleton(topic));
+
+        String tnxId = "txn-" + RandomStringUtils.randomAlphabetic(5);
+        KafkaProducer<Integer, String> producer = buildTransactionProducer(tnxId);
+        producer.initTransactions();
+
+        String baseMsg = "test msg commit - ";
+        producer.beginTransaction();
+        producer.send(new ProducerRecord<>(topic, baseMsg + 0)).get();
+        producer.send(new ProducerRecord<>(topic, baseMsg + 1)).get();
+        producer.flush();
+
+        AtomicInteger messageCount = new AtomicInteger(0);
+        // make sure consumer can't receive unstable messages in `read_committed` mode
+        readAndCheckMessages(consumer, baseMsg, messageCount, isolation.equals("read_committed") ? 0 : 2);
+
+        producer.commitTransaction();
+        producer.beginTransaction();
+        // these two unstable message shouldn't be received in `read_committed` mode
+        producer.send(new ProducerRecord<>(topic, baseMsg + 2)).get();
+        producer.send(new ProducerRecord<>(topic, baseMsg + 3)).get();
+        producer.flush();
+
+        readAndCheckMessages(consumer, baseMsg, messageCount, isolation.equals("read_committed") ? 2 : 4);
+
+        consumer.close();
+        producer.close();
+    }
+
+    private void readAndCheckMessages(KafkaConsumer<Integer, String> consumer, String baseMsg,
+                                      AtomicInteger messageCount, int expectedMessageCount) {
+        while (true) {
+            ConsumerRecords<Integer, String> records = consumer.poll(Duration.ofSeconds(3));
+            if (records.isEmpty()) {
+                break;
+            }
+            for (ConsumerRecord<Integer, String> record : records) {
+                assertEquals(record.value(), baseMsg + messageCount.getAndIncrement());
+            }
+        }
+        // make sure there is no message can be received
+        ConsumerRecords<Integer, String> records = consumer.poll(Duration.ofSeconds(3));
+        assertTrue(records.isEmpty());
+        // make sure only receive the expected number of stable messages
+        assertEquals(messageCount.get(), expectedMessageCount);
+    }
+
 }