This is an automated email from the ASF dual-hosted git repository.
sijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 233c77f Issue 1967: Non-persistent topic drop too many messages
(#1994)
233c77f is described below
commit 233c77fca1ac1bfbc439c2246a45673ff366bfc3
Author: Sijie Guo <[email protected]>
AuthorDate: Wed Jun 20 16:36:50 2018 -0700
Issue 1967: Non-persistent topic drop too many messages (#1994)
*Motivation*
Fixes #1967.
in 2.0 there is an optimization change on grouping acknowledgements.
The [optimization
change](https://github.com/apache/incubator-pulsar/commit/19dd2c502725e3b45e56541576508626c4213091#diff-debb36270f152d2533b03902f214fa9aR674)
avoided delivering messages that are already "delivered". However in
non-persistent topic, pulsar doesn't store entries to bookkeeper, so no ledger
id and entry id
are assigned for those messages. Pulsar uses 0 for both ledger id and
entry. So the optimization change treats messages as already "delivered". so
pulsar client
doesn't dispatch those messages even it already received them from broker.
*Changes*
- Make AcknowledgementsGroupingTracker an interface
- Introduce a no-op implementation for non-persistent topics
Signed-off-by: Sijie Guo <[email protected]>
---
.../impl/AcknowledgmentsGroupingTracker.java | 185 +--------------------
.../apache/pulsar/client/impl/ConsumerImpl.java | 11 +-
...NonPersistentAcknowledgmentGroupingTracker.java | 57 +++++++
... PersistentAcknowledgmentsGroupingTracker.java} | 11 +-
.../impl/AcknowledgementsGroupingTrackerTest.java | 6 +-
.../org/apache/pulsar/common/naming/TopicName.java | 4 +
6 files changed, 86 insertions(+), 188 deletions(-)
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/AcknowledgmentsGroupingTracker.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/AcknowledgmentsGroupingTracker.java
index 68ab1fb..195a9bb 100644
---
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/AcknowledgmentsGroupingTracker.java
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/AcknowledgmentsGroupingTracker.java
@@ -18,193 +18,22 @@
*/
package org.apache.pulsar.client.impl;
-import io.netty.buffer.ByteBuf;
-import io.netty.channel.EventLoopGroup;
-
-import java.io.Closeable;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
import java.util.Map;
-import java.util.concurrent.ConcurrentSkipListSet;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
-
-import lombok.extern.slf4j.Slf4j;
-
-import org.apache.commons.lang3.tuple.Pair;
import org.apache.pulsar.client.api.MessageId;
-import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
-import org.apache.pulsar.common.api.Commands;
import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
/**
- * Group the acknowledgments for a certain time and then sends them out in a
single protobuf command.
+ * Acknowledgments grouping tracker.
*/
-@Slf4j
-public class AcknowledgmentsGroupingTracker implements Closeable {
-
- /**
- * When reaching the max group size, an ack command is sent out immediately
- */
- private static final int MAX_ACK_GROUP_SIZE = 1000;
-
- private final ConsumerImpl<?> consumer;
-
- private final long acknowledgementGroupTimeMicros;
-
- /**
- * Latest cumulative ack sent to broker
- */
- private volatile MessageIdImpl lastCumulativeAck = (MessageIdImpl)
MessageId.earliest;
-
- private static final
AtomicReferenceFieldUpdater<AcknowledgmentsGroupingTracker, MessageIdImpl>
LAST_CUMULATIVE_ACK_UPDATER = AtomicReferenceFieldUpdater
- .newUpdater(AcknowledgmentsGroupingTracker.class,
MessageIdImpl.class, "lastCumulativeAck");
-
- /**
- * This is a set of all the individual acks that the application has
issued and that were not already sent to
- * broker.
- */
- private final ConcurrentSkipListSet<MessageIdImpl> pendingIndividualAcks;
-
- private final ScheduledFuture<?> scheduledTask;
-
- public AcknowledgmentsGroupingTracker(ConsumerImpl<?> consumer,
ConsumerConfigurationData<?> conf,
- EventLoopGroup eventLoopGroup) {
- this.consumer = consumer;
- this.pendingIndividualAcks = new ConcurrentSkipListSet<>();
- this.acknowledgementGroupTimeMicros =
conf.getAcknowledgementsGroupTimeMicros();
-
- if (acknowledgementGroupTimeMicros > 0) {
- scheduledTask =
eventLoopGroup.next().scheduleWithFixedDelay(this::flush,
acknowledgementGroupTimeMicros,
- acknowledgementGroupTimeMicros, TimeUnit.MICROSECONDS);
- } else {
- scheduledTask = null;
- }
- }
-
- /**
- * Since the ack are delayed, we need to do some best-effort duplicate
check to discard messages that are being
- * resent after a disconnection and for which the user has already sent an
acknowlowdgement.
- */
- public boolean isDuplicate(MessageId messageId) {
- if (messageId.compareTo(lastCumulativeAck) <= 0) {
- // Already included in a cumulative ack
- return true;
- } else {
- return pendingIndividualAcks.contains(messageId);
- }
- }
+public interface AcknowledgmentsGroupingTracker extends AutoCloseable {
- public void addAcknowledgment(MessageIdImpl msgId, AckType ackType,
Map<String, Long> properties) {
- if (acknowledgementGroupTimeMicros == 0 || !properties.isEmpty()) {
- // We cannot group acks if the delay is 0 or when there are
properties attached to it. Fortunately that's an
- // uncommon condition since it's only used for the compaction
subscription.
- doImmediateAck(msgId, ackType, properties);
- } else if (ackType == AckType.Cumulative) {
- doCumulativeAck(msgId);
- } else {
- // Individual ack
- pendingIndividualAcks.add(msgId);
- if (pendingIndividualAcks.size() >= MAX_ACK_GROUP_SIZE) {
- flush();
- }
- }
- }
+ boolean isDuplicate(MessageId messageId);
- private void doCumulativeAck(MessageIdImpl msgId) {
- // Handle concurrent updates from different threads
- while (true) {
- MessageIdImpl lastCumlativeAck = this.lastCumulativeAck;
- if (msgId.compareTo(lastCumlativeAck) > 0) {
- if (LAST_CUMULATIVE_ACK_UPDATER.compareAndSet(this,
lastCumlativeAck, msgId)) {
- // Successfully updated the last cumlative ack. Next flush
iteration will send this to broker.
- return;
- }
- } else {
- // message id acknowledging an before the current last
cumulative ack
- return;
- }
- }
- }
+ void addAcknowledgment(MessageIdImpl msgId, AckType ackType, Map<String,
Long> properties);
- private boolean doImmediateAck(MessageIdImpl msgId, AckType ackType,
Map<String, Long> properties) {
- ClientCnx cnx = consumer.getClientCnx();
-
- if (cnx == null) {
- return false;
- }
-
- final ByteBuf cmd = Commands.newAck(consumer.consumerId,
msgId.getLedgerId(), msgId.getEntryId(), ackType, null,
- properties);
-
- cnx.ctx().writeAndFlush(cmd, cnx.ctx().voidPromise());
- return true;
- }
-
- /**
- * Flush all the pending acks and send them to the broker
- */
- public void flush() {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Flushing pending acks to broker:
last-cumulative-ack: {} -- individual-acks: {}", consumer,
- lastCumulativeAck, pendingIndividualAcks);
- }
-
- ClientCnx cnx = consumer.getClientCnx();
-
- if (cnx == null) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Cannot flush pending acks since we're not
connected to broker", consumer);
- }
- return;
- }
-
- if (!lastCumulativeAck.equals(MessageId.earliest)) {
- ByteBuf cmd = Commands.newAck(consumer.consumerId,
lastCumulativeAck.ledgerId, lastCumulativeAck.entryId,
- AckType.Cumulative, null, Collections.emptyMap());
- cnx.ctx().write(cmd, cnx.ctx().voidPromise());
- }
-
- // Flush all individual acks
- if (!pendingIndividualAcks.isEmpty()) {
- if
(Commands.peerSupportsMultiMessageAcknowledgment(cnx.getRemoteEndpointProtocolVersion()))
{
- // We can send 1 single protobuf command with all individual
acks
- List<Pair<Long, Long>> entriesToAck = new
ArrayList<>(pendingIndividualAcks.size());
- while (true) {
- MessageIdImpl msgId = pendingIndividualAcks.pollFirst();
- if (msgId == null) {
- break;
- }
-
- entriesToAck.add(Pair.of(msgId.getLedgerId(),
msgId.getEntryId()));
- }
-
-
cnx.ctx().write(Commands.newMultiMessageAck(consumer.consumerId, entriesToAck),
- cnx.ctx().voidPromise());
- } else {
- // When talking to older brokers, send the acknowledgments
individually
- while (true) {
- MessageIdImpl msgId = pendingIndividualAcks.pollFirst();
- if (msgId == null) {
- break;
- }
-
- cnx.ctx().write(Commands.newAck(consumer.consumerId,
msgId.getLedgerId(), msgId.getEntryId(),
- AckType.Individual, null, Collections.emptyMap()),
cnx.ctx().voidPromise());
- }
- }
- }
-
- cnx.ctx().flush();
- }
+ void flush();
@Override
- public void close() {
- flush();
- if (scheduledTask != null) {
- scheduledTask.cancel(true);
- }
- }
+ void close();
+
}
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
index dc98dc2..dd3b5c4 100644
---
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
@@ -72,6 +72,7 @@ import
org.apache.pulsar.common.api.proto.PulsarApi.MessageMetadata;
import org.apache.pulsar.common.api.proto.PulsarApi.ProtocolVersion;
import org.apache.pulsar.common.compression.CompressionCodec;
import org.apache.pulsar.common.compression.CompressionCodecProvider;
+import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.util.FutureUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -152,7 +153,15 @@ public class ConsumerImpl<T> extends ConsumerBase<T>
implements ConnectionHandle
this.priorityLevel = conf.getPriorityLevel();
this.readCompacted = conf.isReadCompacted();
this.subscriptionInitialPosition =
conf.getSubscriptionInitialPosition();
- this.acknowledgmentsGroupingTracker = new
AcknowledgmentsGroupingTracker(this, conf, client.eventLoopGroup());
+
+ TopicName topicName = TopicName.get(topic);
+ if (topicName.isPersistent()) {
+ this.acknowledgmentsGroupingTracker =
+ new PersistentAcknowledgmentsGroupingTracker(this, conf,
client.eventLoopGroup());
+ } else {
+ this.acknowledgmentsGroupingTracker =
+ NonPersistentAcknowledgmentGroupingTracker.of();
+ }
if (client.getConfiguration().getStatsIntervalSeconds() > 0) {
stats = new ConsumerStatsRecorderImpl(client, conf, this);
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/NonPersistentAcknowledgmentGroupingTracker.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/NonPersistentAcknowledgmentGroupingTracker.java
new file mode 100644
index 0000000..fbf6d03
--- /dev/null
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/NonPersistentAcknowledgmentGroupingTracker.java
@@ -0,0 +1,57 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl;
+
+import java.util.Map;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
+
+/**
+ * A no-op acknowledgment grouping tracker.
+ */
+public class NonPersistentAcknowledgmentGroupingTracker implements
AcknowledgmentsGroupingTracker {
+
+ public static NonPersistentAcknowledgmentGroupingTracker of() {
+ return INSTANCE;
+ }
+
+ private static final NonPersistentAcknowledgmentGroupingTracker INSTANCE =
new NonPersistentAcknowledgmentGroupingTracker();
+
+ private NonPersistentAcknowledgmentGroupingTracker() {}
+
+ @Override
+ public boolean isDuplicate(MessageId messageId) {
+ return false;
+ }
+
+ @Override
+ public void addAcknowledgment(MessageIdImpl msgId, AckType ackType,
Map<String, Long> properties) {
+ // no-op
+ }
+
+ @Override
+ public void flush() {
+ // no-op
+ }
+
+ @Override
+ public void close() {
+ // no-op
+ }
+}
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/AcknowledgmentsGroupingTracker.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java
similarity index 93%
copy from
pulsar-client/src/main/java/org/apache/pulsar/client/impl/AcknowledgmentsGroupingTracker.java
copy to
pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java
index 68ab1fb..0fefe9c 100644
---
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/AcknowledgmentsGroupingTracker.java
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java
@@ -21,7 +21,6 @@ package org.apache.pulsar.client.impl;
import io.netty.buffer.ByteBuf;
import io.netty.channel.EventLoopGroup;
-import java.io.Closeable;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@@ -43,7 +42,7 @@ import
org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
* Group the acknowledgments for a certain time and then sends them out in a
single protobuf command.
*/
@Slf4j
-public class AcknowledgmentsGroupingTracker implements Closeable {
+public class PersistentAcknowledgmentsGroupingTracker implements
AcknowledgmentsGroupingTracker {
/**
* When reaching the max group size, an ack command is sent out immediately
@@ -59,8 +58,8 @@ public class AcknowledgmentsGroupingTracker implements
Closeable {
*/
private volatile MessageIdImpl lastCumulativeAck = (MessageIdImpl)
MessageId.earliest;
- private static final
AtomicReferenceFieldUpdater<AcknowledgmentsGroupingTracker, MessageIdImpl>
LAST_CUMULATIVE_ACK_UPDATER = AtomicReferenceFieldUpdater
- .newUpdater(AcknowledgmentsGroupingTracker.class,
MessageIdImpl.class, "lastCumulativeAck");
+ private static final
AtomicReferenceFieldUpdater<PersistentAcknowledgmentsGroupingTracker,
MessageIdImpl> LAST_CUMULATIVE_ACK_UPDATER = AtomicReferenceFieldUpdater
+ .newUpdater(PersistentAcknowledgmentsGroupingTracker.class,
MessageIdImpl.class, "lastCumulativeAck");
/**
* This is a set of all the individual acks that the application has
issued and that were not already sent to
@@ -70,8 +69,8 @@ public class AcknowledgmentsGroupingTracker implements
Closeable {
private final ScheduledFuture<?> scheduledTask;
- public AcknowledgmentsGroupingTracker(ConsumerImpl<?> consumer,
ConsumerConfigurationData<?> conf,
- EventLoopGroup eventLoopGroup) {
+ public PersistentAcknowledgmentsGroupingTracker(ConsumerImpl<?> consumer,
ConsumerConfigurationData<?> conf,
+ EventLoopGroup
eventLoopGroup) {
this.consumer = consumer;
this.pendingIndividualAcks = new ConcurrentSkipListSet<>();
this.acknowledgementGroupTimeMicros =
conf.getAcknowledgementsGroupTimeMicros();
diff --git
a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/AcknowledgementsGroupingTrackerTest.java
b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/AcknowledgementsGroupingTrackerTest.java
index b6a3d2a..fc6df91 100644
---
a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/AcknowledgementsGroupingTrackerTest.java
+++
b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/AcknowledgementsGroupingTrackerTest.java
@@ -61,7 +61,7 @@ public class AcknowledgementsGroupingTrackerTest {
public void testAckTracker() throws Exception {
ConsumerConfigurationData<?> conf = new ConsumerConfigurationData<>();
conf.setAcknowledgementsGroupTimeMicros(TimeUnit.SECONDS.toMicros(10));
- AcknowledgmentsGroupingTracker tracker = new
AcknowledgmentsGroupingTracker(consumer, conf, eventLoopGroup);
+ PersistentAcknowledgmentsGroupingTracker tracker = new
PersistentAcknowledgmentsGroupingTracker(consumer, conf, eventLoopGroup);
MessageIdImpl msg1 = new MessageIdImpl(5, 1, 0);
MessageIdImpl msg2 = new MessageIdImpl(5, 2, 0);
@@ -119,7 +119,7 @@ public class AcknowledgementsGroupingTrackerTest {
public void testImmediateAckingTracker() throws Exception {
ConsumerConfigurationData<?> conf = new ConsumerConfigurationData<>();
conf.setAcknowledgementsGroupTimeMicros(0);
- AcknowledgmentsGroupingTracker tracker = new
AcknowledgmentsGroupingTracker(consumer, conf, eventLoopGroup);
+ PersistentAcknowledgmentsGroupingTracker tracker = new
PersistentAcknowledgmentsGroupingTracker(consumer, conf, eventLoopGroup);
MessageIdImpl msg1 = new MessageIdImpl(5, 1, 0);
MessageIdImpl msg2 = new MessageIdImpl(5, 2, 0);
@@ -146,7 +146,7 @@ public class AcknowledgementsGroupingTrackerTest {
public void testAckTrackerMultiAck() throws Exception {
ConsumerConfigurationData<?> conf = new ConsumerConfigurationData<>();
conf.setAcknowledgementsGroupTimeMicros(TimeUnit.SECONDS.toMicros(10));
- AcknowledgmentsGroupingTracker tracker = new
AcknowledgmentsGroupingTracker(consumer, conf, eventLoopGroup);
+ PersistentAcknowledgmentsGroupingTracker tracker = new
PersistentAcknowledgmentsGroupingTracker(consumer, conf, eventLoopGroup);
when(cnx.getRemoteEndpointProtocolVersion()).thenReturn(ProtocolVersion.v12_VALUE);
diff --git
a/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java
b/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java
index ba5da90..ef45598 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java
@@ -170,6 +170,10 @@ public class TopicName implements ServiceUnitId {
}
+ public boolean isPersistent() {
+ return TopicDomain.persistent == domain;
+ }
+
/**
* Extract the namespace portion out of a completeTopicName name.
*
