Fix tests

Author: BewareMyPower

pulsar-client-api/src/main/java/org/apache/pulsar/client/api/TopicMessageId.java

@@ -18,6 +18,8 @@
  */
 package org.apache.pulsar.client.api;
 
+import java.util.BitSet;
+
 /**
  * The MessageId used for a consumer that subscribes multiple topics or partitioned topics.
  *
@@ -49,13 +51,13 @@ static TopicMessageId create(String topic, MessageId messageId) {
     /**
      * The simplest implementation of a TopicMessageId interface.
      */
-    class Impl implements TopicMessageId {
+    class Impl implements MessageIdAdv, TopicMessageId {
         private final String topic;
-        private final MessageId messageId;
+        private final MessageIdAdv messageId;
 
         public Impl(String topic, MessageId messageId) {
             this.topic = topic;
-            this.messageId = messageId;
+            this.messageId = (MessageIdAdv) messageId;
         }
 
         @Override
@@ -68,6 +70,41 @@ public String getOwnerTopic() {
             return topic;
         }
 
+        @Override
+        public long getLedgerId() {
+            return messageId.getLedgerId();
+        }
+
+        @Override
+        public long getEntryId() {
+            return messageId.getEntryId();
+        }
+
+        @Override
+        public int getPartitionIndex() {
+            return messageId.getPartitionIndex();
+        }
+
+        @Override
+        public int getBatchIndex() {
+            return messageId.getBatchIndex();
+        }
+
+        @Override
+        public int getBatchSize() {
+            return messageId.getBatchSize();
+        }
+
+        @Override
+        public BitSet getAckSet() {
+            return messageId.getAckSet();
+        }
+
+        @Override
+        public MessageIdAdv getFirstChunkMessageId() {
+            return messageId.getFirstChunkMessageId();
+        }
+
         @Override
         public int compareTo(MessageId o) {
             return messageId.compareTo(o);

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

@@ -2115,7 +2115,7 @@ private CompletableFuture<Void> seekAsyncInternal(long requestId, ByteBuf seek,
         ClientCnx cnx = cnx();
 
         MessageIdAdv originSeekMessageId = seekMessageId;
-        seekMessageId = new BatchMessageIdImpl((MessageIdImpl) seekId);
+        seekMessageId = (MessageIdAdv) seekId;
         duringSeek.set(true);
         log.info("[{}][{}] Seeking subscription to {}", topic, subscription, seekBy);
 

pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java

@@ -182,8 +182,7 @@ public CompletableFuture<Void> addAcknowledgment(MessageId msgId, AckType ackTyp
                                                      Map<String, Long> properties) {
         MessageIdAdv msgIdAdv = (MessageIdAdv) msgId;
         if (MessageIdAdvUtils.isBatch(msgIdAdv)) {
-            BatchMessageIdImpl batchMessageId = (BatchMessageIdImpl) msgId;
-            return addAcknowledgment(MessageIdAdvUtils.discardBatch(msgId), ackType, properties, batchMessageId);
+            return addAcknowledgment(MessageIdAdvUtils.discardBatch(msgId), ackType, properties, msgIdAdv);
         } else {
             return addAcknowledgment(msgIdAdv, ackType, properties, null);
         }
@@ -229,7 +228,7 @@ private CompletableFuture<Void> addAcknowledgment(MessageIdAdv msgId,
                 } else {
                     consumer.onAcknowledgeCumulative(msgId, null);
                 }
-                if (batchMessageId == null || MessageIdAdvUtils.acknowledge(msgId, false)) {
+                if (batchMessageId == null || MessageIdAdvUtils.acknowledge(batchMessageId, false)) {
                     return doCumulativeAck(msgId, properties, null);
                 } else if (batchIndexAckEnabled) {
                     return doCumulativeBatchIndexAck(batchMessageId, properties);