You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2021/12/09 13:52:46 UTC
[pulsar] branch branch-2.8 updated (935f8ba -> 9b8e7d2)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git.
from 935f8ba Cancel scheduled tasks when deleting ManagedLedgerImpl (#12565)
new e466185 [Broker] Fix producer getting incorrectly removed from topic's producers map (#12846)
new 9b8e7d2 [ServerCnx] Close connection after receiving unexpected SendCommand (#12780)
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../pulsar/broker/service/AbstractTopic.java | 8 +-
.../org/apache/pulsar/broker/service/Producer.java | 27 +++----
.../apache/pulsar/broker/service/ServerCnx.java | 4 +-
.../pulsar/broker/service/PersistentTopicTest.java | 14 +++-
.../pulsar/broker/service/ServerCnxTest.java | 88 ++++++++++++++++++++++
5 files changed, 116 insertions(+), 25 deletions(-)
[pulsar] 01/02: [Broker] Fix producer getting incorrectly removed from topic's producers map (#12846)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit e466185794ede3c01ed49e92e07b0bc0dc529928
Author: Michael Marshall <mi...@datastax.com>
AuthorDate: Thu Nov 18 02:42:56 2021 -0600
[Broker] Fix producer getting incorrectly removed from topic's producers map (#12846)
(cherry picked from commit e33687d3f202ab104d41ad086c48b66b6f0d5ff5)
---
.../pulsar/broker/service/AbstractTopic.java | 8 +--
.../org/apache/pulsar/broker/service/Producer.java | 27 ++++-----
.../pulsar/broker/service/PersistentTopicTest.java | 14 ++++-
.../pulsar/broker/service/ServerCnxTest.java | 66 ++++++++++++++++++++++
4 files changed, 91 insertions(+), 24 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
index 5d24147..2576050 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
@@ -650,13 +650,9 @@ public abstract class AbstractTopic implements Topic {
private void tryOverwriteOldProducer(Producer oldProducer, Producer newProducer)
throws BrokerServiceException {
- boolean canOverwrite = false;
- if (oldProducer.equals(newProducer) && !isUserProvidedProducerName(oldProducer)
- && !isUserProvidedProducerName(newProducer) && newProducer.getEpoch() > oldProducer.getEpoch()) {
+ if (newProducer.isSuccessorTo(oldProducer) && !isUserProvidedProducerName(oldProducer)
+ && !isUserProvidedProducerName(newProducer)) {
oldProducer.close(false);
- canOverwrite = true;
- }
- if (canOverwrite) {
if (!producers.replace(newProducer.getProducerName(), oldProducer, newProducer)) {
// Met concurrent update, throw exception here so that client can try reconnect later.
throw new BrokerServiceException.NamingException("Producer with name '" + newProducer.getProducerName()
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java
index d13e2f9..64e0fd6 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java
@@ -138,22 +138,17 @@ public class Producer {
this.clientAddress = cnx.clientSourceAddress();
}
- @Override
- public int hashCode() {
- return Objects.hash(producerName);
- }
-
- @Override
- public boolean equals(Object obj) {
- if (obj instanceof Producer) {
- Producer other = (Producer) obj;
- return Objects.equals(producerName, other.producerName)
- && Objects.equals(topic, other.topic)
- && producerId == other.producerId
- && Objects.equals(cnx, other.cnx);
- }
-
- return false;
+ /**
+ * Method to determine if this producer can replace another producer.
+ * @param other - producer to compare to this one
+ * @return true if this producer is a subsequent instantiation of the same logical producer. Otherwise, false.
+ */
+ public boolean isSuccessorTo(Producer other) {
+ return Objects.equals(producerName, other.producerName)
+ && Objects.equals(topic, other.topic)
+ && producerId == other.producerId
+ && Objects.equals(cnx, other.cnx)
+ && other.getEpoch() < epoch;
}
public void publishMessage(long producerId, long sequenceId, ByteBuf headersAndPayload, long batchSize,
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java
index 1f9fa79..9f21e46 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java
@@ -38,6 +38,7 @@ import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertSame;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
@@ -458,11 +459,20 @@ public class PersistentTopicTest extends MockedBookKeeperTestCase {
// OK
}
- // 4. simple remove producer
+ // 4. Try to remove with unequal producer
+ Producer producerCopy = new Producer(topic, serverCnx, 1 /* producer id */, "prod-name",
+ role, false, null, SchemaVersion.Latest, 0, false,
+ ProducerAccessMode.Shared, Optional.empty());
+ topic.removeProducer(producerCopy);
+ // Expect producer to be in map
+ assertEquals(topic.getProducers().size(), 1);
+ assertSame(topic.getProducers().get(producer.getProducerName()), producer);
+
+ // 5. simple remove producer
topic.removeProducer(producer);
assertEquals(topic.getProducers().size(), 0);
- // 5. duplicate remove
+ // 6. duplicate remove
topic.removeProducer(producer); /* noop */
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java
index 2e2f0f2..7b3d375 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java
@@ -827,6 +827,72 @@ public class ServerCnxTest {
channel.finish();
}
+ @Test(timeOut = 30000)
+ public void testCreateProducerTimeoutThenCreateSameNamedProducerShouldFail() throws Exception {
+ resetChannel();
+ setChannelConnected();
+
+ // Delay the topic creation in a deterministic way
+ CompletableFuture<Runnable> openTopicFuture = new CompletableFuture<>();
+ doAnswer(invocationOnMock -> {
+ openTopicFuture.complete(() -> {
+ ((OpenLedgerCallback) invocationOnMock.getArguments()[2]).openLedgerComplete(ledgerMock, null);
+ });
+ return null;
+ }).when(mlFactoryMock).asyncOpen(matches(".*success.*"), any(ManagedLedgerConfig.class),
+ any(OpenLedgerCallback.class), any(Supplier.class), any());
+
+ // In a create producer timeout from client side we expect to see this sequence of commands :
+ // 1. create producer
+ // 2. close producer (when the timeout is triggered, which may be before the producer was created on the broker
+ // 3. create producer (triggered by reconnection logic)
+ // Then, when another producer is created with the same name, it should fail. Because we only have one
+ // channel here, we just use a different producer id
+
+ // These operations need to be serialized, to allow the last create producer to finally succeed
+ // (There can be more create/close pairs in the sequence, depending on the client timeout
+
+ String producerName = "my-producer";
+
+ ByteBuf createProducer1 = Commands.newProducer(successTopicName, 1 /* producer id */, 1 /* request id */,
+ producerName, Collections.emptyMap());
+ channel.writeInbound(createProducer1);
+
+ ByteBuf closeProducer = Commands.newCloseProducer(1 /* producer id */, 2 /* request id */ );
+ channel.writeInbound(closeProducer);
+
+ ByteBuf createProducer2 = Commands.newProducer(successTopicName, 1 /* producer id */, 3 /* request id */,
+ producerName, Collections.emptyMap());
+ channel.writeInbound(createProducer2);
+
+ // Complete the topic opening: It will make 2nd producer creation successful
+ openTopicFuture.get().run();
+
+ // Close succeeds
+ Object response = getResponse();
+ assertEquals(response.getClass(), CommandSuccess.class);
+ assertEquals(((CommandSuccess) response).getRequestId(), 2);
+
+ // 2nd producer will be successfully created as topic is open by then
+ response = getResponse();
+ assertEquals(response.getClass(), CommandProducerSuccess.class);
+ assertEquals(((CommandProducerSuccess) response).getRequestId(), 3);
+
+ // Send create command after getting the CommandProducerSuccess to ensure correct ordering
+ ByteBuf createProducer3 = Commands.newProducer(successTopicName, 2 /* producer id */, 4 /* request id */,
+ producerName, Collections.emptyMap());
+ channel.writeInbound(createProducer3);
+
+ // 3nd producer will fail
+ response = getResponse();
+ assertEquals(response.getClass(), CommandError.class);
+ assertEquals(((CommandError) response).getRequestId(), 4);
+
+ assertTrue(channel.isActive());
+
+ channel.finish();
+ }
+
@Test(timeOut = 30000, enabled = false)
public void testCreateProducerMultipleTimeouts() throws Exception {
resetChannel();
[pulsar] 02/02: [ServerCnx] Close connection after receiving unexpected SendCommand (#12780)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 9b8e7d223e98e47e13143c69d194fb99140d8f45
Author: Michael Marshall <mi...@datastax.com>
AuthorDate: Sat Nov 13 17:09:39 2021 -0600
[ServerCnx] Close connection after receiving unexpected SendCommand (#12780)
(cherry picked from commit ba5809553344f074c5dce15618a70f3b20d368c7)
---
.../apache/pulsar/broker/service/ServerCnx.java | 4 +++-
.../pulsar/broker/service/ServerCnxTest.java | 22 ++++++++++++++++++++++
2 files changed, 25 insertions(+), 1 deletion(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
index e18f51e..a8d7f23 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
@@ -1343,7 +1343,9 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
CompletableFuture<Producer> producerFuture = producers.get(send.getProducerId());
if (producerFuture == null || !producerFuture.isDone() || producerFuture.isCompletedExceptionally()) {
- log.warn("[{}] Producer had already been closed: {}", remoteAddress, send.getProducerId());
+ log.warn("[{}] Received message, but the producer is not ready : {}. Closing the connection.",
+ remoteAddress, send.getProducerId());
+ close();
return;
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java
index 7b3d375..63347a3 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java
@@ -660,6 +660,28 @@ public class ServerCnxTest {
}
@Test(timeOut = 30000)
+ public void testSendCommandBeforeCreatingProducer() throws Exception {
+ resetChannel();
+ setChannelConnected();
+
+ // test SEND before producer is created
+ MessageMetadata messageMetadata = new MessageMetadata()
+ .setPublishTime(System.currentTimeMillis())
+ .setProducerName("prod-name")
+ .setSequenceId(0);
+ ByteBuf data = Unpooled.buffer(1024);
+
+ ByteBuf clientCommand = ByteBufPair.coalesce(Commands.newSend(1, 0, 1,
+ ChecksumType.None, messageMetadata, data));
+ channel.writeInbound(Unpooled.copiedBuffer(clientCommand));
+ clientCommand.release();
+
+ // Then expect channel to close
+ Awaitility.await().atMost(10, TimeUnit.SECONDS).until(() -> !channel.isActive());
+ channel.finish();
+ }
+
+ @Test(timeOut = 30000)
public void testUseSameProducerName() throws Exception {
resetChannel();
setChannelConnected();