fix kafka plugin

Signed-off-by: Yupeng Fu <[email protected]>

Author: yupeng9

plugins/ingestion-kafka/src/main/java/org/opensearch/plugin/kafka/KafkaPartitionConsumer.java

@@ -125,25 +125,17 @@ public List<ReadResult<KafkaOffset, KafkaMessage>> readNext(
         long maxMessages,
         int timeoutMillis
     ) throws TimeoutException {
-        List<ReadResult<KafkaOffset, KafkaMessage>> records =
-            AccessController.doPrivileged((PrivilegedAction<List<ReadResult<KafkaOffset, KafkaMessage>>>) () -> fetch(offset.getOffset(),
+        List<ReadResult<KafkaOffset, KafkaMessage>> records = AccessController.doPrivileged(
+            (PrivilegedAction<List<ReadResult<KafkaOffset, KafkaMessage>>>) () -> fetch(
+                offset.getOffset(),
                 includeStart,
                 maxMessages,
                 timeoutMillis
-            ));
+            )
+        );
         return records;
     }
 
-    @Override
-    public KafkaOffset nextPointer() {
-        return new KafkaOffset(lastFetchedOffset + 1);
-    }
-
-    @Override
-    public KafkaOffset nextPointer(KafkaOffset pointer) {
-        return new KafkaOffset(pointer.getOffset() + 1);
-    }
-
     @Override
     public IngestionShardPointer earliestPointer() {
         long startOffset = AccessController.doPrivileged(

plugins/ingestion-kafka/src/test/java/org/opensearch/plugin/kafka/KafkaPartitionConsumerTests.java

@@ -65,7 +65,6 @@ public void testReadNext() throws Exception {
 
         assertEquals(1, result.size());
         assertEquals("message", new String(result.get(0).getMessage().getPayload(), StandardCharsets.UTF_8));
-        assertEquals(1, consumer.nextPointer().getOffset());
         assertEquals(0, consumer.getShardId());
         assertEquals("client1", consumer.getClientId());
     }