diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java index 9babd2ad419ec..1197e36a5da19 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java @@ -37,7 +37,8 @@ public class ConnectionHandler { protected final Backoff backoff; private static final AtomicLongFieldUpdater 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; protected volatile long lastConnectionClosedTimestamp = 0L; interface Connection { @@ -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 { @@ -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); @@ -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); } @@ -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) { diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java index 40c9b99987f1a..a37da16af3a0c 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java @@ -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; } @@ -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; @@ -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) { @@ -1363,9 +1363,17 @@ public Iterator 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()); @@ -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(); @@ -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(); @@ -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) { @@ -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); } }); } @@ -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 @@ -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 msgIterator = pendingMessages.iterator(); OpSendMsg pendingRegisteringOp = null; @@ -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); } }