pulsar-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From si...@apache.org
Subject [incubator-pulsar] branch branch-2.1 updated: [compaction] make topic compaction works with partitioned topic (#2367)
Date Mon, 27 Aug 2018 18:18:39 GMT
This is an automated email from the ASF dual-hosted git repository.

sijie pushed a commit to branch branch-2.1
in repository https://gitbox.apache.org/repos/asf/incubator-pulsar.git


The following commit(s) were added to refs/heads/branch-2.1 by this push:
     new f97a8f7  [compaction] make topic compaction works with partitioned topic (#2367)
f97a8f7 is described below

commit f97a8f7d136ddc609d01ce54e20f45e9949b04df
Author: Sijie Guo <guosijie@gmail.com>
AuthorDate: Tue Aug 14 11:02:24 2018 -0700

    [compaction] make topic compaction works with partitioned topic (#2367)
    
    * [compaction] make topic compaction works with partitioned topic
    
     ### Motivation
    
    Topic compaction doesn't work with partitioned topic.
    
     ### Changes
    
    - make `RawReaderImpl` and `ReaderImpl` return message with partition idx
    - make broker service `Consumer` deliver MessageIdData with partition idx
    - add an integration test to ensure compaction work with partitioned topic
---
 .../java/org/apache/pulsar/broker/service/Consumer.java  |  9 +++++++--
 .../org/apache/pulsar/client/impl/RawReaderImpl.java     | 16 ++++++++++++----
 .../org/apache/pulsar/compaction/TwoPhaseCompactor.java  |  8 ++++----
 .../PersistentDispatcherFailoverConsumerTest.java        |  2 +-
 .../java/org/apache/pulsar/client/impl/ReaderImpl.java   |  4 +++-
 5 files changed, 27 insertions(+), 12 deletions(-)

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java
index c3befa5..4c54018 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java
@@ -73,6 +73,7 @@ public class Consumer {
     private final String appId;
     private AuthenticationDataSource authenticationData;
     private final String topicName;
+    private final int partitionIdx;
     private final InitialPosition subscriptionInitialPosition;
 
     private final long consumerId;
@@ -119,6 +120,7 @@ public class Consumer {
         this.subscription = subscription;
         this.subType = subType;
         this.topicName = topicName;
+        this.partitionIdx = TopicName.getPartitionIndex(topicName);
         this.consumerId = consumerId;
         this.priorityLevel = priorityLevel;
         this.readCompacted = readCompacted;
@@ -239,8 +241,11 @@ public class Consumer {
                 Entry entry = entries.get(i);
                 PositionImpl pos = (PositionImpl) entry.getPosition();
                 MessageIdData.Builder messageIdBuilder = MessageIdData.newBuilder();
-                MessageIdData messageId = messageIdBuilder.setLedgerId(pos.getLedgerId()).setEntryId(pos.getEntryId())
-                        .build();
+                MessageIdData messageId = messageIdBuilder
+                    .setLedgerId(pos.getLedgerId())
+                    .setEntryId(pos.getEntryId())
+                    .setPartition(partitionIdx)
+                    .build();
 
                 ByteBuf metadataAndPayload = entry.getDataBuffer();
                 // increment ref-count of data and release at the end of process: so, we
can get chance to call entry.release
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
index e768c3e..ae1a4db 100644
--- 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
@@ -35,6 +35,7 @@ import org.apache.pulsar.client.api.SubscriptionType;
 import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
 import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
 import org.apache.pulsar.common.api.proto.PulsarApi.MessageIdData;
+import org.apache.pulsar.common.naming.TopicName;
 import org.apache.pulsar.common.util.collections.GrowableArrayBlockingQueue;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -102,8 +103,15 @@ public class RawReaderImpl implements RawReader {
 
         RawConsumerImpl(PulsarClientImpl client, ConsumerConfigurationData<byte[]>
conf,
                 CompletableFuture<Consumer<byte[]>> consumerFuture) {
-            super(client, conf.getSingleTopic(), conf, client.externalExecutorProvider().getExecutor(),
-1,
-                    consumerFuture, SubscriptionMode.Durable, MessageId.earliest, Schema.BYTES);
+            super(client,
+                conf.getSingleTopic(),
+                conf,
+                client.externalExecutorProvider().getExecutor(),
+                TopicName.getPartitionIndex(conf.getSingleTopic()),
+                consumerFuture,
+                SubscriptionMode.Durable,
+                MessageId.earliest,
+                Schema.BYTES);
             incomingRawMessages = new GrowableArrayBlockingQueue<>();
             pendingRawReceives = new ConcurrentLinkedQueue<>();
         }
@@ -172,8 +180,8 @@ public class RawReaderImpl implements RawReader {
         @Override
         void messageReceived(MessageIdData messageId, ByteBuf headersAndPayload, ClientCnx
cnx) {
             if (log.isDebugEnabled()) {
-                log.debug("[{}][{}] Received raw message: {}/{}", topic, subscription,
-                          messageId.getLedgerId(), messageId.getEntryId());
+                log.debug("[{}][{}] Received raw message: {}/{}/{}", topic, subscription,
+                          messageId.getEntryId(), messageId.getLedgerId(), messageId.getPartition());
             }
             incomingRawMessages.add(
                     new RawMessageAndCnx(new RawMessageImpl(messageId, headersAndPayload),
cnx));
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/TwoPhaseCompactor.java
b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/TwoPhaseCompactor.java
index b4ee68a..cc3f710 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/TwoPhaseCompactor.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/TwoPhaseCompactor.java
@@ -146,11 +146,11 @@ public class TwoPhaseCompactor extends Compactor {
 
     private void scheduleTimeout(CompletableFuture<RawMessage> future) {
         Future<?> timeout = scheduler.schedule(() -> {
-                future.completeExceptionally(new TimeoutException("Timeout"));
-            }, 10, TimeUnit.SECONDS);
+            future.completeExceptionally(new TimeoutException("Timeout"));
+        }, 10, TimeUnit.SECONDS);
         future.whenComplete((res, exception) -> {
-                timeout.cancel(true);
-            });
+            timeout.cancel(true);
+        });
     }
 
     private CompletableFuture<Long> phaseTwo(RawReader reader, MessageId from, MessageId
to,
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java
index ccd32e1..6f16733 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java
@@ -579,7 +579,7 @@ public class PersistentDispatcherFailoverConsumerTest {
 
     private Consumer createConsumer(int priority, int permit, boolean blocked, int id) throws
Exception {
         Consumer consumer =
-                new Consumer(null, SubType.Shared, null, id, priority, ""+id, 5000,
+                new Consumer(null, SubType.Shared, "test-topic", id, priority, ""+id, 5000,
                         serverCnx, "appId", Collections.emptyMap(), false /* read compacted
*/, InitialPosition.Latest);
         try {
             consumer.flowPermits(permit);
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderImpl.java
index 00f8af0..aafd125 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderImpl.java
@@ -36,6 +36,7 @@ import org.apache.pulsar.client.api.SubscriptionType;
 import org.apache.pulsar.client.impl.ConsumerImpl.SubscriptionMode;
 import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
 import org.apache.pulsar.client.impl.conf.ReaderConfigurationData;
+import org.apache.pulsar.common.naming.TopicName;
 
 public class ReaderImpl<T> implements Reader<T> {
 
@@ -82,8 +83,9 @@ public class ReaderImpl<T> implements Reader<T> {
             consumerConfiguration.setCryptoKeyReader(readerConfiguration.getCryptoKeyReader());
         }
 
+        final int partitionIdx = TopicName.getPartitionIndex(readerConfiguration.getTopicName());
         consumer = new ConsumerImpl<>(client, readerConfiguration.getTopicName(), consumerConfiguration,
listenerExecutor,
-                -1, consumerFuture, SubscriptionMode.NonDurable, readerConfiguration.getStartMessageId(),
schema);
+                partitionIdx, consumerFuture, SubscriptionMode.NonDurable, readerConfiguration.getStartMessageId(),
schema);
     }
 
     @Override


Mime
View raw message