Skip to content

Commit

Permalink
Fix timeout issue in MsgFactory & thread-safety issue in MsgSender/Re…
Browse files Browse the repository at this point in the history 
…ceiver

* fix connection open/close timeout paths
* use non ForkJoinPool variants while composing completable futures (Azure#4)
* fix thread-safety in MessageSender & MessageReceiver in error cases(Azure#4)
  • Loading branch information
@SreeramGarlapati
SreeramGarlapati authored Apr 6, 2017
1 parent b378b50 commit 017d24b
Show file tree
Hide file tree
Showing 8 changed files with 207 additions and 107 deletions.
14 changes: 7 additions & 7 deletions azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/EventHubClient.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -132,7 +132,7 @@ public static CompletableFuture<EventHubClient> createFromConnectionString(final
final EventHubClient eventHubClient = new EventHubClient(connStr);

return MessagingFactory.createFromConnectionString(connectionString.toString(), retryPolicy)
.thenApplyAsync(new Function<MessagingFactory, EventHubClient>() {
.thenApply(new Function<MessagingFactory, EventHubClient>() {
@Override
public EventHubClient apply(MessagingFactory factory) {
eventHubClient.underlyingFactory = factory;
Expand Down Expand Up @@ -205,7 +205,7 @@ public final CompletableFuture<Void> send(final EventData data) {
throw new IllegalArgumentException("EventData cannot be empty.");
}

return this.createInternalSender().thenComposeAsync(new Function<Void, CompletableFuture<Void>>() {
return this.createInternalSender().thenCompose(new Function<Void, CompletableFuture<Void>>() {
@Override
public CompletableFuture<Void> apply(Void voidArg) {
return EventHubClient.this.sender.send(data.toAmqpMessage());
Expand Down Expand Up @@ -295,7 +295,7 @@ public final CompletableFuture<Void> send(final Iterable<EventData> eventDatas)
throw new IllegalArgumentException("Empty batch of EventData cannot be sent.");
}

return this.createInternalSender().thenComposeAsync(new Function<Void, CompletableFuture<Void>>() {
return this.createInternalSender().thenCompose(new Function<Void, CompletableFuture<Void>>() {
@Override
public CompletableFuture<Void> apply(Void voidArg) {
return EventHubClient.this.sender.send(EventDataUtil.toAmqpMessages(eventDatas));
Expand Down Expand Up @@ -371,7 +371,7 @@ public final CompletableFuture<Void> send(final EventData eventData, final Strin
throw new IllegalArgumentException("partitionKey cannot be null");
}

return this.createInternalSender().thenComposeAsync(new Function<Void, CompletableFuture<Void>>() {
return this.createInternalSender().thenCompose(new Function<Void, CompletableFuture<Void>>() {
@Override
public CompletableFuture<Void> apply(Void voidArg) {
return EventHubClient.this.sender.send(eventData.toAmqpMessage(partitionKey));
Expand Down Expand Up @@ -445,7 +445,7 @@ public final CompletableFuture<Void> send(final Iterable<EventData> eventDatas,
String.format(Locale.US, "PartitionKey exceeds the maximum allowed length of partitionKey: {0}", ClientConstants.MAX_PARTITION_KEY_LENGTH));
}

return this.createInternalSender().thenComposeAsync(new Function<Void, CompletableFuture<Void>>() {
return this.createInternalSender().thenCompose(new Function<Void, CompletableFuture<Void>>() {
@Override
public CompletableFuture<Void> apply(Void voidArg) {
return EventHubClient.this.sender.send(EventDataUtil.toAmqpMessages(eventDatas, partitionKey));
Expand Down Expand Up @@ -1222,7 +1222,7 @@ public CompletableFuture<Void> onClose() {
if (this.underlyingFactory != null) {
synchronized (this.senderCreateSync) {
final CompletableFuture<Void> internalSenderClose = this.sender != null
? this.sender.close().thenComposeAsync(new Function<Void, CompletableFuture<Void>>() {
? this.sender.close().thenCompose(new Function<Void, CompletableFuture<Void>>() {
@Override
public CompletableFuture<Void> apply(Void voidArg) {
return EventHubClient.this.underlyingFactory.close();
Expand All @@ -1242,7 +1242,7 @@ private CompletableFuture<Void> createInternalSender() {
synchronized (this.senderCreateSync) {
if (!this.isSenderCreateStarted) {
this.createSender = MessageSender.create(this.underlyingFactory, StringUtil.getRandomString(), this.eventHubName)
.thenAcceptAsync(new Consumer<MessageSender>() {
.thenAccept(new Consumer<MessageSender>() {
public void accept(MessageSender a) {
EventHubClient.this.sender = a;
}
Expand Down
4 changes: 2 additions & 2 deletions azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/PartitionReceiver.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -124,7 +124,7 @@ static CompletableFuture<PartitionReceiver> create(MessagingFactory factory,
}

final PartitionReceiver receiver = new PartitionReceiver(factory, eventHubName, consumerGroupName, partitionId, startingOffset, offsetInclusive, dateTime, epoch, isEpochReceiver, receiverOptions);
return receiver.createInternalReceiver().thenApplyAsync(new Function<Void, PartitionReceiver>() {
return receiver.createInternalReceiver().thenApply(new Function<Void, PartitionReceiver>() {
public PartitionReceiver apply(Void a) {
return receiver;
}
Expand All @@ -136,7 +136,7 @@ private CompletableFuture<Void> createInternalReceiver() throws ServiceBusExcept
StringUtil.getRandomString(),
String.format("%s/ConsumerGroups/%s/Partitions/%s", this.eventHubName, this.consumerGroupName, this.partitionId),
PartitionReceiver.DEFAULT_PREFETCH_COUNT, this)
.thenAcceptAsync(new Consumer<MessageReceiver>() {
.thenAccept(new Consumer<MessageReceiver>() {
public void accept(MessageReceiver r) {
PartitionReceiver.this.internalReceiver = r;
}
Expand Down
4 changes: 2 additions & 2 deletions azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/PartitionSender.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,7 @@ private PartitionSender(MessagingFactory factory, String eventHubName, String pa
static CompletableFuture<PartitionSender> Create(MessagingFactory factory, String eventHubName, String partitionId) throws ServiceBusException {
final PartitionSender sender = new PartitionSender(factory, eventHubName, partitionId);
return sender.createInternalSender()
.thenApplyAsync(new Function<Void, PartitionSender>() {
.thenApply(new Function<Void, PartitionSender>() {
public PartitionSender apply(Void a) {
return sender;
}
Expand All @@ -47,7 +47,7 @@ public PartitionSender apply(Void a) {
private CompletableFuture<Void> createInternalSender() throws ServiceBusException {
return MessageSender.create(this.factory, StringUtil.getRandomString(),
String.format("%s/Partitions/%s", this.eventHubName, this.partitionId))
.thenAcceptAsync(new Consumer<MessageSender>() {
.thenAccept(new Consumer<MessageSender>() {
public void accept(MessageSender a) {
PartitionSender.this.internalSender = a;
}
Expand Down
30 changes: 18 additions & 12 deletions azure-eventhubs/src/main/java/com/microsoft/azure/servicebus/CBSChannel.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@

import org.apache.qpid.proton.Proton;
import org.apache.qpid.proton.amqp.messaging.ApplicationProperties;
import org.apache.qpid.proton.engine.Session;
import org.apache.qpid.proton.message.Message;
import org.apache.qpid.proton.amqp.messaging.AmqpValue;
import org.apache.qpid.proton.amqp.transport.ErrorCondition;
Expand Down Expand Up @@ -100,21 +101,26 @@ private class OpenRequestResponseChannel implements IOperation<RequestResponseCh
@Override
public void run(IOperationResult<RequestResponseChannel, Exception> operationCallback) {

final Session session = CBSChannel.this.sessionProvider.getSession(
"cbs-session",
null,
new BiConsumer<ErrorCondition, Exception>() {
@Override
public void accept(ErrorCondition error, Exception exception) {
if (error != null)
operationCallback.onError(new AmqpException(error));
else if (exception != null)
operationCallback.onError(exception);
}
});

if (session == null)
return;

final RequestResponseChannel requestResponseChannel = new RequestResponseChannel(
"cbs",
ClientConstants.CBS_ADDRESS,
CBSChannel.this.sessionProvider.getSession(
"cbs-session",
null,
new BiConsumer<ErrorCondition, Exception>() {
@Override
public void accept(ErrorCondition error, Exception exception) {
if (error != null)
operationCallback.onError(new AmqpException(error));
else if (exception != null)
operationCallback.onError(exception);
}
}));
session);

requestResponseChannel.open(
new IOperationResult<Void, Exception>() {
Expand Down
8 changes: 6 additions & 2 deletions azure-eventhubs/src/main/java/com/microsoft/azure/servicebus/ClientEntity.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -102,9 +102,13 @@ public final void closeSync() throws ServiceBusException {
}
}

protected final void throwIfClosed(Throwable cause) {
protected final void throwIfClosed() {
if (this.getIsClosingOrClosed()) {
throw new IllegalStateException(String.format(Locale.US, "Operation not allowed after the %s instance is Closed.", this.getClass().getName()), cause);
throw new IllegalStateException(String.format(Locale.US, "Operation not allowed after the %s instance is Closed.", this.getClass().getName()), this.getLastKnownError());
}
}

protected Exception getLastKnownError() {
return null;
}
}
80 changes: 53 additions & 27 deletions azure-eventhubs/src/main/java/com/microsoft/azure/servicebus/MessageReceiver.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,7 @@ public final class MessageReceiver extends ClientEntity implements IAmqpReceiver
private final ConcurrentLinkedQueue<Message> prefetchedMessages;
private final ReceiveWork receiveWork;
private final CreateAndReceive createAndReceive;
private final Object errorConditionLock;

private int prefetchCount;
private Receiver receiveLink;
Expand Down Expand Up @@ -95,6 +96,7 @@ private MessageReceiver(final MessagingFactory factory,
this.linkOpen = new WorkItem<>(new CompletableFuture<>(), factory.getOperationTimeout());

this.pendingReceives = new ConcurrentLinkedQueue<>();
this.errorConditionLock = new Object();

// onOperationTimeout delegate - per receive call
this.onOperationTimedout = new Runnable() {
Expand Down Expand Up @@ -250,7 +252,7 @@ public void setReceiveTimeout(final Duration value) {
}

public CompletableFuture<Collection<Message>> receive(final int maxMessageCount) {
this.throwIfClosed(this.lastKnownLinkError);
this.throwIfClosed();

if (maxMessageCount <= 0 || maxMessageCount > this.prefetchCount) {
throw new IllegalArgumentException(String.format(Locale.US, "parameter 'maxMessageCount' should be a positive number and should be less than prefetchCount(%s)", this.prefetchCount));
Expand Down Expand Up @@ -289,7 +291,9 @@ public void onOpenComplete(Exception exception) {
this.openTimer.cancel(false);
}

this.lastKnownLinkError = null;
synchronized (this.errorConditionLock) {
this.lastKnownLinkError = null;
}

this.underlyingFactory.getRetryPolicy().resetRetryCount(this.underlyingFactory.getClientId());

Expand All @@ -308,7 +312,9 @@ public void onOpenComplete(Exception exception) {
this.openTimer.cancel(false);
}

this.lastKnownLinkError = exception;
synchronized (this.errorConditionLock) {
this.lastKnownLinkError = exception;
}
}
}

Expand All @@ -330,11 +336,6 @@ public void onReceiveComplete(Delivery delivery) {
this.receiveWork.onEvent();
}

public void onError(final ErrorCondition error) {
final Exception completionException = ExceptionUtil.toException(error);
this.onError(completionException);
}

@Override
public void onError(final Exception exception) {
this.prefetchedMessages.clear();
Expand All @@ -358,7 +359,9 @@ public void onError(final Exception exception) {

this.linkClose.complete(null);
} else {
this.lastKnownLinkError = exception == null ? this.lastKnownLinkError : exception;
synchronized (this.errorConditionLock) {
this.lastKnownLinkError = exception == null ? this.lastKnownLinkError : exception;
}

final Exception completionException = exception == null
? new ServiceBusException(true, "Client encountered transient error for unknown reasons, please retry the operation.") : exception;
Expand Down Expand Up @@ -452,15 +455,17 @@ public void accept(Session session) {

receiver.open();

MessageReceiver.this.receiveLink = receiver;
synchronized (MessageReceiver.this.errorConditionLock) {
MessageReceiver.this.receiveLink = receiver;
}
}
};

final BiConsumer<ErrorCondition, Exception> onSessionOpenFailed = new BiConsumer<ErrorCondition, Exception>() {
@Override
public void accept(ErrorCondition t, Exception u) {
if (t != null)
onError(t);
onError((t != null && t.getCondition() != null) ? ExceptionUtil.toException(t) : null);
else if (u != null)
onError(u);
}
Expand Down Expand Up @@ -526,12 +531,19 @@ private void scheduleLinkOpenTimeout(final TimeoutTracker timeout) {
new Runnable() {
public void run() {
if (!linkOpen.getWork().isDone()) {
Exception operationTimedout = new TimeoutException(
String.format(Locale.US, "%s operation on ReceiveLink(%s) to path(%s) timed out at %s.", "Open", MessageReceiver.this.receiveLink.getName(), MessageReceiver.this.receivePath, ZonedDateTime.now()),
MessageReceiver.this.lastKnownLinkError);
final Receiver link;
final Exception lastReportedLinkError;
synchronized (errorConditionLock) {
link = MessageReceiver.this.receiveLink;
lastReportedLinkError = MessageReceiver.this.lastKnownLinkError;
}

final Exception operationTimedout = new TimeoutException(
String.format(Locale.US, "%s operation on ReceiveLink(%s) to path(%s) timed out at %s.", "Open", link.getName(), MessageReceiver.this.receivePath, ZonedDateTime.now()),
lastReportedLinkError);
if (TRACE_LOGGER.isLoggable(Level.WARNING)) {
TRACE_LOGGER.log(Level.WARNING,
String.format(Locale.US, "receiverPath[%s], linkName[%s], %s call timedout", MessageReceiver.this.receivePath, MessageReceiver.this.receiveLink.getName(), "Open"),
String.format(Locale.US, "receiverPath[%s], linkName[%s], %s call timedout", MessageReceiver.this.receivePath, link.getName(), "Open"),
operationTimedout);
}

Expand All @@ -549,10 +561,15 @@ private void scheduleLinkCloseTimeout(final TimeoutTracker timeout) {
new Runnable() {
public void run() {
if (!linkClose.isDone()) {
Exception operationTimedout = new TimeoutException(String.format(Locale.US, "%s operation on Receive Link(%s) timed out at %s", "Close", MessageReceiver.this.receiveLink.getName(), ZonedDateTime.now()));
final Receiver link;
synchronized (errorConditionLock) {
link = MessageReceiver.this.receiveLink;
}

final Exception operationTimedout = new TimeoutException(String.format(Locale.US, "%s operation on Receive Link(%s) timed out at %s", "Close", link.getName(), ZonedDateTime.now()));
if (TRACE_LOGGER.isLoggable(Level.WARNING)) {
TRACE_LOGGER.log(Level.WARNING,
String.format(Locale.US, "receiverPath[%s], linkName[%s], %s call timedout", MessageReceiver.this.receivePath, MessageReceiver.this.receiveLink.getName(), "Close"),
String.format(Locale.US, "receiverPath[%s], linkName[%s], %s call timedout", MessageReceiver.this.receivePath, link.getName(), "Close"),
operationTimedout);
}

Expand All @@ -567,25 +584,27 @@ public void run() {

@Override
public void onClose(ErrorCondition condition) {
if (condition == null || condition.getCondition() == null) {
this.onError((Exception) null);
} else {
this.onError(condition);
}
final Exception completionException = (condition != null && condition.getCondition() != null) ? ExceptionUtil.toException(condition) : null;
this.onError(completionException);
}

@Override
public ErrorContext getContext() {
final Receiver link;
synchronized (this.errorConditionLock) {
link = this.receiveLink;
}

final boolean isLinkOpened = this.linkOpen != null && this.linkOpen.getWork().isDone();
final String referenceId = this.receiveLink != null && this.receiveLink.getRemoteProperties() != null && this.receiveLink.getRemoteProperties().containsKey(ClientConstants.TRACKING_ID_PROPERTY)
? this.receiveLink.getRemoteProperties().get(ClientConstants.TRACKING_ID_PROPERTY).toString()
: ((this.receiveLink != null) ? this.receiveLink.getName() : null);
final String referenceId = link != null && link.getRemoteProperties() != null && link.getRemoteProperties().containsKey(ClientConstants.TRACKING_ID_PROPERTY)
? link.getRemoteProperties().get(ClientConstants.TRACKING_ID_PROPERTY).toString()
: ((link != null) ? link.getName() : null);

ReceiverContext errorContext = new ReceiverContext(this.underlyingFactory != null ? this.underlyingFactory.getHostName() : null,
final ReceiverContext errorContext = new ReceiverContext(this.underlyingFactory != null ? this.underlyingFactory.getHostName() : null,
this.receivePath,
referenceId,
isLinkOpened ? this.prefetchCount : null,
isLinkOpened && this.receiveLink != null ? this.receiveLink.getCredit() : null,
isLinkOpened && link != null ? link.getCredit() : null,
isLinkOpened && this.prefetchedMessages != null ? this.prefetchedMessages.size() : null);

return errorContext;
Expand Down Expand Up @@ -628,6 +647,13 @@ public void onEvent() {
return this.linkClose;
}

@Override
protected Exception getLastKnownError() {
synchronized (this.errorConditionLock) {
return this.lastKnownLinkError;
}
}

private final class ReceiveWork extends DispatchHandler {

@Override
Expand Down
Loading

0 comments on commit 017d24b

Please sign in to comment.