Skip to content

[Java Client] Use epoch to version producer's cnx to prevent early delivery of messages #12779

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

Merged
eolivelli merged 2 commits into from
Dec 7, 2021
Merged

[Java Client] Use epoch to version producer's cnx to prevent early delivery of messages #12779

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
5931afd
[Java Client] Use epoch to version producer's cnx to prevent early de…
michaeljmarshall Nov 11, 2021
1c1791c
Update initial epoch value: it now gets incremented before the first …
michaeljmarshall Nov 13, 2021
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
20 changes: 13 additions & 7 deletions pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,8 @@ public class ConnectionHandler {
protected final Backoff backoff;
private static final AtomicLongFieldUpdater<ConnectionHandler> EPOCH_UPDATER = AtomicLongFieldUpdater
.newUpdater(ConnectionHandler.class, "epoch");
private volatile long epoch = 0L;
// Start with -1L because it gets incremented before sending on the first connection
private volatile long epoch = -1L;

@Jason918 Jason918 Nov 13, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Does the starting value of epoch mean anything? In my understanding, it's just used for comparison, right?

@michaeljmarshall michaeljmarshall Nov 14, 2021

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't believe the starting value inherently means anything. However, the test ProducerCreationTest#testGeneratedNameProducerReconnect asserts that the epoch value is 2 after a single producer reconnect. I could have updated the test or the starting value. I chose to update the starting value here to maintain the original behavior.

@Jason918 Jason918 Nov 14, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

OK, I see. I am OK with this new starting value.
But IMHO, it's better to assert that the epoch value increased after a single producer reconnect to avoid more flaky test.

protected volatile long lastConnectionClosedTimestamp = 0L;

interface Connection {
Expand Down Expand Up @@ -106,7 +107,6 @@ protected void reconnectLater(Throwable exception) {
if (state.changeToConnecting()) {
state.client.timer().newTimeout(timeout -> {
log.info("[{}] [{}] Reconnecting after connection was closed", state.topic, state.getHandlerName());
incrementEpoch();
grabCnx();
}, delayMs, TimeUnit.MILLISECONDS);
} else {
Expand All @@ -115,10 +115,6 @@ protected void reconnectLater(Throwable exception) {
}
}

protected long incrementEpoch() {
return EPOCH_UPDATER.incrementAndGet(this);
}

public void connectionClosed(ClientCnx cnx) {
lastConnectionClosedTimestamp = System.currentTimeMillis();
state.client.getCnxPool().releaseConnection(cnx);
Expand All @@ -133,7 +129,6 @@ public void connectionClosed(ClientCnx cnx) {
delayMs / 1000.0);
state.client.timer().newTimeout(timeout -> {
log.info("[{}] [{}] Reconnecting after timeout", state.topic, state.getHandlerName());
incrementEpoch();
grabCnx();
}, delayMs, TimeUnit.MILLISECONDS);
}
Expand All @@ -151,6 +146,17 @@ protected void setClientCnx(ClientCnx clientCnx) {
CLIENT_CNX_UPDATER.set(this, clientCnx);
}

/**
* Update the {@link ClientCnx} for the class, then increment and get the epoch value. Note that the epoch value is
* currently only used by the {@link ProducerImpl}.
* @param clientCnx - the new {@link ClientCnx}
* @return the epoch value to use for this pair of {@link ClientCnx} and {@link ProducerImpl}
*/
protected long switchClientCnx(ClientCnx clientCnx) {
setClientCnx(clientCnx);
return EPOCH_UPDATER.incrementAndGet(this);
}

private boolean isValidStateForReconnection() {
State state = this.state.getState();
switch (state) {
Expand Down
43 changes: 30 additions & 13 deletions pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -631,7 +631,7 @@ private boolean rePopulateMessageSchema(MessageImpl msg) {
return true;
}

private void tryRegisterSchema(ClientCnx cnx, MessageImpl msg, SendCallback callback) {
private void tryRegisterSchema(ClientCnx cnx, MessageImpl msg, SendCallback callback, long expectedCnxEpoch) {
if (!changeToRegisteringSchemaState()) {
return;
}
Expand All @@ -656,7 +656,7 @@ private void tryRegisterSchema(ClientCnx cnx, MessageImpl msg, SendCallback call
}
cnx.ctx().channel().eventLoop().execute(() -> {
synchronized (ProducerImpl.this) {
recoverProcessOpSendMsgFrom(cnx, msg);
recoverProcessOpSendMsgFrom(cnx, msg, expectedCnxEpoch);
}
});
return null;
Expand Down Expand Up @@ -1088,7 +1088,7 @@ protected synchronized void recoverChecksumError(ClientCnx cnx, long sequenceId)
}
}
// as msg is not corrupted : let producer resend pending-messages again including checksum failed message
resendMessages(cnx);
resendMessages(cnx, this.connectionHandler.getEpoch());
}

protected synchronized void recoverNotAllowedError(long sequenceId) {
Expand Down Expand Up @@ -1363,9 +1363,17 @@ public Iterator<OpSendMsg> iterator() {
public void connectionOpened(final ClientCnx cnx) {
previousExceptions.clear();

// we set the cnx reference before registering the producer on the cnx, so if the cnx breaks before creating the
// producer, it will try to grab a new cnx
connectionHandler.setClientCnx(cnx);
final long epoch;
synchronized (this) {
// Because the state could have been updated while retrieving the connection, we set it back to connecting,
// as long as the change from current state to connecting is a valid state change.
if (!changeToConnecting()) {
return;
}
// We set the cnx reference before registering the producer on the cnx, so if the cnx breaks before creating
// the producer, it will try to grab a new cnx. We also increment and get the epoch value for the producer.
epoch = connectionHandler.switchClientCnx(cnx);
}
cnx.registerProducer(producerId, this);

log.info("[{}] [{}] Creating producer on cnx {}", topic, producerName, cnx.ctx().channel());
Expand Down Expand Up @@ -1402,7 +1410,7 @@ public void connectionOpened(final ClientCnx cnx) {

cnx.sendRequestWithId(
Commands.newProducer(topic, producerId, requestId, producerName, conf.isEncryptionEnabled(), metadata,
schemaInfo, connectionHandler.getEpoch(), userProvidedProducerName,
schemaInfo, epoch, userProvidedProducerName,
conf.getAccessMode(), topicEpoch, client.conf.isEnableTransaction()),
requestId).thenAccept(response -> {
String producerName = response.getProducerName();
Expand Down Expand Up @@ -1464,7 +1472,7 @@ public void connectionOpened(final ClientCnx cnx) {
}
}, 0, conf.getBatchingMaxPublishDelayMicros(), TimeUnit.MICROSECONDS);
}
resendMessages(cnx);
resendMessages(cnx, epoch);
}
}).exceptionally((e) -> {
Throwable cause = e.getCause();
Expand Down Expand Up @@ -1568,7 +1576,7 @@ public void connectionFailed(PulsarClientException exception) {
}
}

private void resendMessages(ClientCnx cnx) {
private void resendMessages(ClientCnx cnx, long expectedEpoch) {
cnx.ctx().channel().eventLoop().execute(() -> {
synchronized (this) {
if (getState() == State.Closing || getState() == State.Closed) {
Expand All @@ -1595,7 +1603,7 @@ private void resendMessages(ClientCnx cnx) {
}

log.info("[{}] [{}] Re-Sending {} messages to server", topic, producerName, messagesToResend);
recoverProcessOpSendMsgFrom(cnx, null);
recoverProcessOpSendMsgFrom(cnx, null, expectedEpoch);
}
});
}
Expand Down Expand Up @@ -1824,7 +1832,7 @@ protected void processOpSendMsg(OpSendMsg op) {
if (shouldWriteOpSendMsg()) {
ClientCnx cnx = cnx();
if (op.msg != null && op.msg.getSchemaState() == None) {
tryRegisterSchema(cnx, op.msg, op.callback);
tryRegisterSchema(cnx, op.msg, op.callback, this.connectionHandler.getEpoch());
return;
}
// If we do have a connection, the message is sent immediately, otherwise we'll try again once a new
Expand Down Expand Up @@ -1855,7 +1863,16 @@ protected boolean shouldWriteOpSendMsg() {
return isConnected();
}

private void recoverProcessOpSendMsgFrom(ClientCnx cnx, MessageImpl from) {
// Must acquire a lock on ProducerImpl.this before calling method.
private void recoverProcessOpSendMsgFrom(ClientCnx cnx, MessageImpl from, long expectedEpoch) {
if (expectedEpoch != this.connectionHandler.getEpoch() || cnx() == null) {
// In this case, the cnx passed to this method is no longer the active connection. This method will get
// called again once the new connection registers the producer with the broker.
log.info("[{}][{}] Producer epoch mismatch or the current connection is null. Skip re-sending the "
+ " {} pending messages since they will deliver using another connection.", topic, producerName,
pendingMessages.size());
return;
}
final boolean stripChecksum = cnx.getRemoteEndpointProtocolVersion() < brokerChecksumSupportedVersion();
Iterator<OpSendMsg> msgIterator = pendingMessages.iterator();
OpSendMsg pendingRegisteringOp = null;
Expand Down Expand Up @@ -1904,7 +1921,7 @@ private void recoverProcessOpSendMsgFrom(ClientCnx cnx, MessageImpl from) {
return;
}
if (pendingRegisteringOp != null) {
tryRegisterSchema(cnx, pendingRegisteringOp.msg, pendingRegisteringOp.callback);
tryRegisterSchema(cnx, pendingRegisteringOp.msg, pendingRegisteringOp.callback, expectedEpoch);
}
}

Expand Down