[Broker] Optimize RawReader#create when using Compactor (apache#14447)

(cherry picked from commit 992dfbd) (cherry picked from commit d9edbef)
datastax · Apr 5, 2022 · 1dc8e60 · 1dc8e60
1 parent c7f5307
commit 1dc8e60
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 1 deletion.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/client/api/RawReader.java b/pulsar-broker/src/main/java/org/apache/pulsar/client/api/RawReader.java
@@ -34,7 +34,7 @@ public interface RawReader {
     static CompletableFuture<RawReader> create(PulsarClient client, String topic, String subscription) {
         CompletableFuture<Consumer<byte[]>> future = new CompletableFuture<>();
         RawReader r = new RawReaderImpl((PulsarClientImpl) client, topic, subscription, future);
-        return future.thenCompose(x -> x.seekAsync(MessageId.earliest)).thenApply(__ -> r);
+        return future.thenApply(__ -> r);
     }
 
     /**

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java
@@ -31,6 +31,7 @@
 import org.apache.pulsar.client.api.RawMessage;
 import org.apache.pulsar.client.api.RawReader;
 import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
 import org.apache.pulsar.client.api.SubscriptionType;
 import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
 import org.apache.pulsar.common.api.proto.CommandAck.AckType;
@@ -56,6 +57,7 @@ public RawReaderImpl(PulsarClientImpl client, String topic, String subscription,
         consumerConfiguration.setSubscriptionType(SubscriptionType.Exclusive);
         consumerConfiguration.setReceiverQueueSize(DEFAULT_RECEIVER_QUEUE_SIZE);
         consumerConfiguration.setReadCompacted(true);
+        consumerConfiguration.setSubscriptionInitialPosition(SubscriptionInitialPosition.Earliest);
 
         consumer = new RawConsumerImpl(client, consumerConfiguration,
                                        consumerFuture);