From 17add4b1e88dc6999d0930c906eb026d07345a93 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Sun, 29 Sep 2024 12:24:48 +0800 Subject: [PATCH 1/5] Fix refreshAsync never complete when the topic is empty --- .../apache/pulsar/client/impl/TableViewTest.java | 3 +++ .../apache/pulsar/client/impl/TableViewImpl.java | 13 ++++++++++++- 2 files changed, 15 insertions(+), 1 deletion(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TableViewTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TableViewTest.java index 61ab4de8a3294..5448751160a9c 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TableViewTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TableViewTest.java @@ -173,6 +173,9 @@ public void testRefreshAPI(int partition) throws Exception { TableView tv = pulsarClient.newTableView(Schema.BYTES) .topic(topic) .create(); + // Verify refresh can handle the case when the topic is empty + tv.refreshAsync().get(3, TimeUnit.SECONDS); + // 2. Add a listen action to provide the test environment. // The listen action will be triggered when there are incoming messages every time. // This is a sync operation, so sleep in the listen action can slow down the reading rate of messages. diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TableViewImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TableViewImpl.java index 4f52060497864..464d4d70fa564 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TableViewImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TableViewImpl.java @@ -36,6 +36,7 @@ import org.apache.pulsar.client.api.CryptoKeyReader; import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.MessageId; +import org.apache.pulsar.client.api.MessageIdAdv; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.Reader; import org.apache.pulsar.client.api.ReaderBuilder; @@ -260,6 +261,10 @@ private void handleMessage(Message msg) { public CompletableFuture refreshAsync() { CompletableFuture completableFuture = new CompletableFuture<>(); reader.thenCompose(reader -> getLastMessageIds(reader).thenAccept(lastMessageIds -> { + if (lastMessageIds.isEmpty()) { + completableFuture.complete(null); + return; + } // After get the response of lastMessageIds, put the future and result into `refreshMap` // and then filter out partitions that has been read to the lastMessageID. pendingRefreshRequests.put(completableFuture, lastMessageIds); @@ -292,6 +297,10 @@ private CompletableFuture readAllExistingMessages(Reader reader) { CompletableFuture future = new CompletableFuture<>(); getLastMessageIds(reader).thenAccept(maxMessageIds -> { + if (maxMessageIds.isEmpty()) { + future.complete(null); + return; + } readAllExistingMessages(reader, future, startTime, messagesRead, maxMessageIds); }).exceptionally(ex -> { future.completeExceptionally(ex); @@ -304,7 +313,9 @@ private CompletableFuture> getLastMessageIds(Reader< return reader.getLastMessageIdsAsync().thenApply(lastMessageIds -> { Map maxMessageIds = new ConcurrentHashMap<>(); lastMessageIds.forEach(topicMessageId -> { - maxMessageIds.put(topicMessageId.getOwnerTopic(), topicMessageId); + if (((MessageIdAdv) topicMessageId).getEntryId() >= 0) { + maxMessageIds.put(topicMessageId.getOwnerTopic(), topicMessageId); + } }); return maxMessageIds; }); From 2595616668d1df2f2a7e2a49dd78ed506ac6b1f0 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Sun, 29 Sep 2024 12:35:24 +0800 Subject: [PATCH 2/5] Add BrokerRegistryMetadataStoreIntegrationTest --- .../BrokerRegistryIntegrationTest.java | 7 +++- ...rRegistryMetadataStoreIntegrationTest.java | 33 +++++++++++++++++++ 2 files changed, 39 insertions(+), 1 deletion(-) create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryMetadataStoreIntegrationTest.java diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryIntegrationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryIntegrationTest.java index d6615a8a5b49b..c8dc2692613ed 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryIntegrationTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryIntegrationTest.java @@ -63,10 +63,15 @@ protected void setup() throws Exception { @AfterClass(alwaysRun = true) protected void cleanup() throws Exception { + final var startMs = System.currentTimeMillis(); if (pulsar != null) { pulsar.close(); } + final var elapsedMs = System.currentTimeMillis() - startMs; bk.stop(); + if (elapsedMs > 5000) { + throw new RuntimeException("Broker took " + elapsedMs + "ms to close"); + } } @Test(enabled = false) @@ -105,7 +110,7 @@ public void testRegisterAgain() throws Exception { }); } - private ServiceConfiguration brokerConfig() { + protected ServiceConfiguration brokerConfig() { final var config = new ServiceConfiguration(); config.setClusterName(clusterName); config.setAdvertisedAddress("localhost"); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryMetadataStoreIntegrationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryMetadataStoreIntegrationTest.java new file mode 100644 index 0000000000000..e8cd3043f560f --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryMetadataStoreIntegrationTest.java @@ -0,0 +1,33 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.broker.loadbalance.extensions; + +import org.apache.pulsar.broker.ServiceConfiguration; +import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateMetadataStoreTableViewImpl; + +public class BrokerRegistryMetadataStoreIntegrationTest extends BrokerRegistryIntegrationTest { + + @Override + protected ServiceConfiguration brokerConfig() { + final var config = super.brokerConfig(); + config.setLoadManagerServiceUnitStateTableViewClassName( + ServiceUnitStateMetadataStoreTableViewImpl.class.getName()); + return config; + } +} From 24624f081a2bf26fcdc395052e4f9bdf28023831 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Sun, 29 Sep 2024 15:27:39 +0800 Subject: [PATCH 3/5] Move registerAsync to another thread --- .../loadbalance/extensions/BrokerRegistryImpl.java | 14 ++++++-------- 1 file changed, 6 insertions(+), 8 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryImpl.java index 9fd0518a054cc..a13b332e6eb5f 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryImpl.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryImpl.java @@ -83,13 +83,6 @@ public BrokerRegistryImpl(PulsarService pulsar) { this.brokerLookupDataMetadataCache = pulsar.getLocalMetadataStore().getMetadataCache(BrokerLookupData.class); this.scheduler = pulsar.getLoadManagerExecutor(); this.listeners = new ArrayList<>(); - // The registered node is an ephemeral node that could be deleted when the metadata store client's session - // is expired. In this case, we should register again. - this.listeners.add((broker, notificationType) -> { - if (notificationType == NotificationType.Deleted && getBrokerId().equals(broker)) { - registerAsync(); - } - }); this.brokerIdKeyPath = keyPath(pulsar.getBrokerId()); this.brokerLookupData = new BrokerLookupData( pulsar.getWebServiceAddress(), @@ -223,11 +216,16 @@ private void handleMetadataStoreNotification(Notification t) { if (log.isDebugEnabled()) { log.debug("Handle notification: [{}]", t); } + // The registered node is an ephemeral node that could be deleted when the metadata store client's session + // is expired. In this case, we should register again. + final var brokerId = t.getPath().substring(LOADBALANCE_BROKERS_ROOT.length() + 1); + if (t.getType() == NotificationType.Deleted && getBrokerId().equals(brokerId)) { + registerAsync(); + } if (listeners.isEmpty()) { return; } this.scheduler.submit(() -> { - String brokerId = t.getPath().substring(LOADBALANCE_BROKERS_ROOT.length() + 1); for (BiConsumer listener : listeners) { listener.accept(brokerId, t.getType()); } From 09092ea2019716f3ac9e6dbf0a063e746e819ceb Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Sun, 29 Sep 2024 15:32:10 +0800 Subject: [PATCH 4/5] Enable the BrokerRegistryIntegrationTest again --- .../extensions/BrokerRegistryIntegrationTest.java | 6 +++--- .../BrokerRegistryMetadataStoreIntegrationTest.java | 2 ++ 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryIntegrationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryIntegrationTest.java index c8dc2692613ed..232088afb94fe 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryIntegrationTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryIntegrationTest.java @@ -37,7 +37,7 @@ import org.testng.annotations.Test; @Slf4j -@Test(groups = "flaky") +@Test(groups = "broker") public class BrokerRegistryIntegrationTest { private static final String clusterName = "test"; @@ -74,7 +74,7 @@ protected void cleanup() throws Exception { } } - @Test(enabled = false) + @Test public void testRecoverFromNodeDeletion() throws Exception { // Simulate the case that the node was somehow deleted (e.g. by session timeout) Awaitility.await().atMost(Duration.ofSeconds(3)).untilAsserted(() -> Assert.assertEquals( @@ -93,7 +93,7 @@ public void testRecoverFromNodeDeletion() throws Exception { Assert.assertEquals(brokerRegistry.getAvailableBrokersAsync().get(), List.of(pulsar.getBrokerId())); } - @Test(enabled = false) + @Test public void testRegisterAgain() throws Exception { Awaitility.await().atMost(Duration.ofSeconds(3)).untilAsserted(() -> Assert.assertEquals( brokerRegistry.getAvailableBrokersAsync().join(), List.of(pulsar.getBrokerId()))); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryMetadataStoreIntegrationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryMetadataStoreIntegrationTest.java index e8cd3043f560f..3e01b1fad0f21 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryMetadataStoreIntegrationTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryMetadataStoreIntegrationTest.java @@ -20,7 +20,9 @@ import org.apache.pulsar.broker.ServiceConfiguration; import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateMetadataStoreTableViewImpl; +import org.testng.annotations.Test; +@Test(groups = "broker") public class BrokerRegistryMetadataStoreIntegrationTest extends BrokerRegistryIntegrationTest { @Override From 26edb49e7a51f26690a46869e90918c1e2d745ba Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Mon, 30 Sep 2024 10:00:52 +0800 Subject: [PATCH 5/5] Change the method name and add some comments --- .../pulsar/client/impl/TableViewImpl.java | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TableViewImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TableViewImpl.java index 464d4d70fa564..17b49828eeced 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TableViewImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TableViewImpl.java @@ -260,7 +260,7 @@ private void handleMessage(Message msg) { @Override public CompletableFuture refreshAsync() { CompletableFuture completableFuture = new CompletableFuture<>(); - reader.thenCompose(reader -> getLastMessageIds(reader).thenAccept(lastMessageIds -> { + reader.thenCompose(reader -> getLastMessageIdOfNonEmptyTopics(reader).thenAccept(lastMessageIds -> { if (lastMessageIds.isEmpty()) { completableFuture.complete(null); return; @@ -296,12 +296,12 @@ private CompletableFuture readAllExistingMessages(Reader reader) { AtomicLong messagesRead = new AtomicLong(); CompletableFuture future = new CompletableFuture<>(); - getLastMessageIds(reader).thenAccept(maxMessageIds -> { - if (maxMessageIds.isEmpty()) { + getLastMessageIdOfNonEmptyTopics(reader).thenAccept(lastMessageIds -> { + if (lastMessageIds.isEmpty()) { future.complete(null); return; } - readAllExistingMessages(reader, future, startTime, messagesRead, maxMessageIds); + readAllExistingMessages(reader, future, startTime, messagesRead, lastMessageIds); }).exceptionally(ex -> { future.completeExceptionally(ex); return null; @@ -309,15 +309,15 @@ private CompletableFuture readAllExistingMessages(Reader reader) { return future; } - private CompletableFuture> getLastMessageIds(Reader reader) { + private CompletableFuture> getLastMessageIdOfNonEmptyTopics(Reader reader) { return reader.getLastMessageIdsAsync().thenApply(lastMessageIds -> { - Map maxMessageIds = new ConcurrentHashMap<>(); + Map lastMessageIdMap = new ConcurrentHashMap<>(); lastMessageIds.forEach(topicMessageId -> { if (((MessageIdAdv) topicMessageId).getEntryId() >= 0) { - maxMessageIds.put(topicMessageId.getOwnerTopic(), topicMessageId); - } + lastMessageIdMap.put(topicMessageId.getOwnerTopic(), topicMessageId); + } // else: a negative entry id represents an empty topic so that we don't have to read messages from it }); - return maxMessageIds; + return lastMessageIdMap; }); }