Skip to content

Deferred messages for consumers #3155

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
lovelle wants to merge 2 commits into from
Closed

Deferred messages for consumers #3155

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
fe35208
Deferred messages for consumers
lovelle Nov 27, 2018
4ef0ebe
Tests for Delay messages on consumers
lovelle Jan 3, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
165 changes: 154 additions & 11 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,18 +29,13 @@
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.stream.Collectors;

import lombok.Data;

import io.netty.util.concurrent.ScheduledFuture;
import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.impl.PositionImpl;
Expand All @@ -67,6 +62,8 @@
* A Consumer is a consumer currently connected and associated with a Subscription
*/
public class Consumer {
private static final int MAX_REDELIVERY_AT_ONCE = 100;
private static final int SCHED_CHECK_DEFERRED_MS = 1000;
private final Subscription subscription;
private final SubType subType;
private final ServerCnx cnx;
Expand All @@ -76,6 +73,7 @@ public class Consumer {
private final int partitionIdx;
private final InitialPosition subscriptionInitialPosition;

private final long receiverDelay;
private final long consumerId;
private final int priorityLevel;
private final boolean readCompacted;
Expand Down Expand Up @@ -108,14 +106,18 @@ public class Consumer {

private final Map<String, String> metadata;

private final DelayQueue<DelayPositionInfo> delayedPositions;
private volatile ScheduledFuture<?> schedProcessDelay;

public interface SendListener {
void sendComplete(ChannelFuture future, SendMessageInfo sendMessageInfo);
}

public Consumer(Subscription subscription, SubType subType, String topicName, long consumerId,
int priorityLevel, String consumerName,
int maxUnackedMessages, ServerCnx cnx, String appId,
Map<String, String> metadata, boolean readCompacted, InitialPosition subscriptionInitialPosition) throws BrokerServiceException {
Map<String, String> metadata, boolean readCompacted, InitialPosition subscriptionInitialPosition,
long receiverDelay) throws BrokerServiceException {

this.subscription = subscription;
this.subType = subType;
Expand All @@ -132,6 +134,7 @@ public Consumer(Subscription subscription, SubType subType, String topicName, lo
this.msgRedeliver = new Rate();
this.appId = appId;
this.authenticationData = cnx.authenticationData;
this.receiverDelay = receiverDelay;
PERMITS_RECEIVED_WHILE_CONSUMER_BLOCKED_UPDATER.set(this, 0);
MESSAGE_PERMITS_UPDATER.set(this, 0);
UNACKED_MESSAGES_UPDATER.set(this, 0);
Expand All @@ -151,6 +154,66 @@ public Consumer(Subscription subscription, SubType subType, String topicName, lo
// We don't need to keep track of pending acks if the subscription is not shared
this.pendingAcks = null;
}

if (isReceiverDelayEnabled()) {
this.delayedPositions = new DelayQueue<>();
this.schedProcessDelay = ctx().channel().eventLoop().schedule(
this::processDelayEntries, SCHED_CHECK_DEFERRED_MS, TimeUnit.MILLISECONDS);
} else {
this.delayedPositions = null;
this.schedProcessDelay = null;
}
}

private void processDelayEntries() {
if (delayedPositions.isEmpty()) {
log.debug("[{}-{}] no pending messages to deliver for deferred consumer", topicName, subscription);
this.schedProcessDelay = ctx().channel().eventLoop().schedule(
this::processDelayEntries, SCHED_CHECK_DEFERRED_MS, TimeUnit.MILLISECONDS);
return;
}

List<PositionImpl> redeliverPositions = new ArrayList<>();
DelayPositionInfo entry;
int totalRedeliveryMessages = 0;
boolean hasBeenScheduled = false;
for (int i = 0; (entry = delayedPositions.peek()) != null && i <= MAX_REDELIVERY_AT_ONCE; i++) {
final long delay = entry.getDelay(TimeUnit.MILLISECONDS);

if (delay > 0) {
this.schedProcessDelay = ctx().channel().eventLoop().schedule(this::processDelayEntries,
Math.min(delay, SCHED_CHECK_DEFERRED_MS), TimeUnit.MILLISECONDS);
hasBeenScheduled = true;
break;
}

totalRedeliveryMessages += entry.getBatchSize();
redeliverPositions.add(entry.getPosition());
delayedPositions.remove();
}

if (!redeliverPositions.isEmpty()) {
log.debug("[{}-{}] redelivering {} messages on delayed consumer and available permits are: {}",
topicName, subscription, totalRedeliveryMessages, getAvailablePermits());

addAndGetUnAckedMsgs(this, -totalRedeliveryMessages);
blockedConsumerOnUnackedMsgs = false;
subscription.redeliverUnacknowledgedMessages(this, redeliverPositions);
msgRedeliver.recordMultipleEvents(totalRedeliveryMessages, totalRedeliveryMessages);

PERMITS_RECEIVED_WHILE_CONSUMER_BLOCKED_UPDATER.getAndAdd(this, -totalRedeliveryMessages);
MESSAGE_PERMITS_UPDATER.getAndAdd(this, totalRedeliveryMessages);
subscription.consumerFlow(this, totalRedeliveryMessages);
}

if (!hasBeenScheduled) {
if (delayedPositions.isEmpty()) {
this.schedProcessDelay = ctx().channel().eventLoop().schedule(
this::processDelayEntries, SCHED_CHECK_DEFERRED_MS, TimeUnit.MILLISECONDS);
} else {
ctx().channel().eventLoop().execute(this::processDelayEntries);
}
}
}

public SubType subType() {
Expand Down Expand Up @@ -201,10 +264,11 @@ public SendMessageInfo sendMessages(final List<Entry> entries) {
*
* @return a SendMessageInfo object that contains the detail of what was sent to consumer
*/
public SendMessageInfo sendMessages(final List<Entry> entries, SendListener listener) {
public SendMessageInfo sendMessages(final List<Entry> e, SendListener listener) {
final ChannelHandlerContext ctx = cnx.ctx();
final SendMessageInfo sentMessages = new SendMessageInfo();
final ChannelPromise writePromise = listener != null ? ctx.newPromise() : ctx.voidPromise();
final CopyOnWriteArrayList<Entry> entries = new CopyOnWriteArrayList<>(e);

if (listener != null) {
writePromise.addListener(future -> listener.sendComplete(writePromise, sentMessages));
Expand Down Expand Up @@ -303,6 +367,7 @@ public static int getBatchSizeforEntry(ByteBuf metadataAndPayload, Subscription
}

void updatePermitsAndPendingAcks(final List<Entry> entries, SendMessageInfo sentMessages) throws PulsarServerException {
final List<Entry> entriesToDiscard = new ArrayList<>();
int permitsToReduce = 0;
Iterator<Entry> iter = entries.iterator();
boolean unsupportedVersion = false;
Expand All @@ -314,12 +379,26 @@ void updatePermitsAndPendingAcks(final List<Entry> entries, SendMessageInfo sent
int batchSize = getBatchSizeforEntry(metadataAndPayload, subscription, consumerId);
if (batchSize == -1) {
// this would suggest that the message might have been corrupted
iter.remove();
entriesToDiscard.add(entry);
PositionImpl pos = (PositionImpl) entry.getPosition();
entry.release();
subscription.acknowledgeMessage(Collections.singletonList(pos), AckType.Individual, Collections.emptyMap());
continue;
}

if (isReceiverDelayEnabled()) {
final long timeToDeliver = readPublishFrom(metadataAndPayload) + receiverDelay;

if (isExpired(timeToDeliver)) {
entriesToDiscard.add(entry);
PositionImpl pos = (PositionImpl) entry.getPosition();
permitsToReduce += batchSize;
entry.release();
delayedPositions.add(new DelayPositionInfo(pos, timeToDeliver, batchSize));
continue;
}
}

if (pendingAcks != null) {
pendingAcks.put(entry.getLedgerId(), entry.getEntryId(), batchSize, 0);
}
Expand All @@ -343,11 +422,32 @@ void updatePermitsAndPendingAcks(final List<Entry> entries, SendMessageInfo sent
}
}

if (!entriesToDiscard.isEmpty()) {
entries.removeAll(entriesToDiscard);
}

msgOut.recordMultipleEvents(permitsToReduce, totalReadableBytes);
sentMessages.totalSentMessages = permitsToReduce;
sentMessages.totalSentMessageBytes = totalReadableBytes;
}

private static boolean isExpired(long timeToDeliver) {
return System.currentTimeMillis() < timeToDeliver;
}

private static long readPublishFrom(final ByteBuf metadataAndPayload) {
metadataAndPayload.markReaderIndex();
final PulsarApi.MessageMetadata metadata = Commands.parseMessageMetadata(metadataAndPayload);
metadataAndPayload.resetReaderIndex();
long publishTime = metadata.getPublishTime();
metadata.recycle();
return publishTime;
}

private boolean isReceiverDelayEnabled() {
return receiverDelay > 0;
}

public boolean isWritable() {
return cnx.isWritable();
}
Expand All @@ -361,12 +461,23 @@ public void sendError(ByteBuf error) {
* pending message acks
*/
public void close() throws BrokerServiceException {
tearDownDeferredData();
subscription.removeConsumer(this);
cnx.removedConsumer(this);
}

private void tearDownDeferredData() {
if (isReceiverDelayEnabled()) {
if (!schedProcessDelay.isDone()) {
schedProcessDelay.cancel(false);
}
delayedPositions.clear();
}
}

public void disconnect() {
log.info("Disconnecting consumer: {}", this);
tearDownDeferredData();
cnx.closeConsumer(this);
try {
close();
Expand All @@ -377,6 +488,7 @@ public void disconnect() {

void doUnsubscribe(final long requestId) {
final ChannelHandlerContext ctx = cnx.ctx();
tearDownDeferredData();

subscription.doUnsubscribe(this).thenAccept(v -> {
log.info("Unsubscribed successfully from {}", subscription);
Expand Down Expand Up @@ -698,5 +810,36 @@ public void setTotalSentMessageBytes(long totalSentMessageBytes) {
}
}

private static final class DelayPositionInfo implements Delayed {

private final PositionImpl position;
private final long timeToDeliver;
private final int batchSize;

DelayPositionInfo(PositionImpl position, long timeToDeliver, int batchSize) {
this.position = position;
this.batchSize = batchSize;
this.timeToDeliver = timeToDeliver;
}

PositionImpl getPosition() {
return position;
}

int getBatchSize() {
return batchSize;
}

@Override
public long getDelay(TimeUnit unit) {
return unit.toMillis(timeToDeliver - System.currentTimeMillis());
}

@Override
public int compareTo(Delayed d) {
return Long.compare(timeToDeliver, ((DelayPositionInfo) d).timeToDeliver);
}
}

private static final Logger log = LoggerFactory.getLogger(Consumer.class);
}
5 changes: 3 additions & 2 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -527,6 +527,7 @@ protected void handleSubscribe(final CommandSubscribe subscribe) {
final Map<String, String> metadata = CommandUtils.metadataFromCommand(subscribe);
final InitialPosition initialPosition = subscribe.getInitialPosition();
final SchemaData schema = subscribe.hasSchema() ? getSchema(subscribe.getSchema()) : null;
final long receiverDelay = subscribe.hasReceiverDelay() ? subscribe.getReceiverDelay() : 0;

CompletableFuture<Boolean> isProxyAuthorizedFuture;
if (service.isAuthorizationEnabled() && originalPrincipal != null) {
Expand Down Expand Up @@ -596,7 +597,7 @@ protected void handleSubscribe(final CommandSubscribe subscribe) {
return topic.subscribe(ServerCnx.this, subscriptionName, consumerId,
subType, priorityLevel, consumerName, isDurable,
startMessageId, metadata,
readCompacted, initialPosition);
readCompacted, initialPosition, receiverDelay);
} else {
return FutureUtil.failedFuture(
new IncompatibleSchemaException(
Expand All @@ -607,7 +608,7 @@ protected void handleSubscribe(final CommandSubscribe subscribe) {
} else {
return topic.subscribe(ServerCnx.this, subscriptionName, consumerId,
subType, priorityLevel, consumerName, isDurable,
startMessageId, metadata, readCompacted, initialPosition);
startMessageId, metadata, readCompacted, initialPosition, receiverDelay);
}
})
.thenAccept(consumer -> {
Expand Down
2 changes: 1 addition & 1 deletion pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,7 +85,7 @@ default long getOriginalSequenceId() {

CompletableFuture<Consumer> subscribe(ServerCnx cnx, String subscriptionName, long consumerId, SubType subType,
int priorityLevel, String consumerName, boolean isDurable, MessageId startMessageId,
Map<String, String> metadata, boolean readCompacted, InitialPosition initialPosition);
Map<String, String> metadata, boolean readCompacted, InitialPosition initialPosition, long receiverDelay);

CompletableFuture<Subscription> createSubscription(String subscriptionName, InitialPosition initialPosition);

Expand Down
4 changes: 2 additions & 2 deletions ...oker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -309,7 +309,7 @@ public void removeProducer(Producer producer) {
@Override
public CompletableFuture<Consumer> subscribe(final ServerCnx cnx, String subscriptionName, long consumerId,
SubType subType, int priorityLevel, String consumerName, boolean isDurable, MessageId startMessageId,
Map<String, String> metadata, boolean readCompacted, InitialPosition initialPosition) {
Map<String, String> metadata, boolean readCompacted, InitialPosition initialPosition, long receiverDelay) {

final CompletableFuture<Consumer> future = new CompletableFuture<>();

Expand Down Expand Up @@ -360,7 +360,7 @@ public CompletableFuture<Consumer> subscribe(final ServerCnx cnx, String subscri

try {
Consumer consumer = new Consumer(subscription, subType, topic, consumerId, priorityLevel, consumerName, 0, cnx,
cnx.getRole(), metadata, readCompacted, initialPosition);
cnx.getRole(), metadata, readCompacted, initialPosition, receiverDelay);
subscription.addConsumer(consumer);
if (!cnx.isActive()) {
consumer.close();
Expand Down
4 changes: 2 additions & 2 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -482,7 +482,7 @@ public void removeProducer(Producer producer) {
@Override
public CompletableFuture<Consumer> subscribe(final ServerCnx cnx, String subscriptionName, long consumerId,
SubType subType, int priorityLevel, String consumerName, boolean isDurable, MessageId startMessageId,
Map<String, String> metadata, boolean readCompacted, InitialPosition initialPosition) {
Map<String, String> metadata, boolean readCompacted, InitialPosition initialPosition, long receiverDelay) {

final CompletableFuture<Consumer> future = new CompletableFuture<>();

Expand Down Expand Up @@ -557,7 +557,7 @@ public CompletableFuture<Consumer> subscribe(final ServerCnx cnx, String subscri
subscriptionFuture.thenAccept(subscription -> {
try {
Consumer consumer = new Consumer(subscription, subType, topic, consumerId, priorityLevel, consumerName,
maxUnackedMessages, cnx, cnx.getRole(), metadata, readCompacted, initialPosition);
maxUnackedMessages, cnx, cnx.getRole(), metadata, readCompacted, initialPosition, receiverDelay);
subscription.addConsumer(consumer);
if (!cnx.isActive()) {
consumer.close();
Expand Down
Loading