From 04db1163b6cddb11b79dc25c81995e03bfee21db Mon Sep 17 00:00:00 2001
From: dockerzhang <dockerzhang@apache.org>
Date: Mon, 21 Dec 2020 11:57:24 +0800
Subject: [PATCH] using the first entry, index 0 as base offset

---
 .../pulsar/handlers/kop/utils/MessageRecordUtils.java          | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/MessageRecordUtils.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/MessageRecordUtils.java
index 525b4486cc..cd4ec237ad 100644
--- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/MessageRecordUtils.java
+++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/MessageRecordUtils.java
@@ -339,7 +339,8 @@ public static MemoryRecords entriesToRecords(List<org.apache.bookkeeper.mledger.
             MemoryRecordsBuilder builder = new MemoryRecordsBuilder(outputStream, magic,
                 org.apache.kafka.common.record.CompressionType.NONE,
                 TimestampType.CREATE_TIME,
-                MessageIdUtils.getOffset(entries.get(0).getLedgerId(), 0),
+                // using the first entry, index 0 as base offset
+                MessageIdUtils.getOffset(entries.get(0).getLedgerId(), entries.get(0).getEntryId(), 0),
                 RecordBatch.NO_TIMESTAMP,
                 RecordBatch.NO_PRODUCER_ID,
                 RecordBatch.NO_PRODUCER_EPOCH,