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 2024/10/30 11:41:07 UTC
(pulsar) branch branch-3.0 updated (eddf3956318 -> b0af444a419)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from eddf3956318 [fix] [broker] Fix race-condition causing repeated delete topic (#23522)
new b7b69dcf817 [fix][client] Fix Reader.hasMessageAvailable return wrong value after seeking by timestamp with startMessageIdInclusive (#23502)
new b0af444a419 [fix][standalone] correctly delete bookie registration znode (#23497)
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/zookeeper/LocalBookkeeperEnsemble.java | 26 ++++++++------
.../org/apache/pulsar/client/impl/ReaderTest.java | 40 ++++++++++++++++++++++
.../apache/pulsar/client/impl/ConsumerImpl.java | 2 ++
3 files changed, 57 insertions(+), 11 deletions(-)
(pulsar) 02/02: [fix][standalone] correctly delete bookie registration znode (#23497)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit b0af444a419dbd678dd35e24e2e054c832fe7e11
Author: Zixuan Liu <no...@gmail.com>
AuthorDate: Fri Oct 25 11:24:57 2024 +0800
[fix][standalone] correctly delete bookie registration znode (#23497)
Signed-off-by: Zixuan Liu <no...@gmail.com>
(cherry picked from commit ebb3cb5384d188b3c9d6b3fe89c2f66254103bf7)
---
.../pulsar/zookeeper/LocalBookkeeperEnsemble.java | 26 +++++++++++++---------
1 file changed, 15 insertions(+), 11 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/zookeeper/LocalBookkeeperEnsemble.java b/pulsar-broker/src/main/java/org/apache/pulsar/zookeeper/LocalBookkeeperEnsemble.java
index a1b0b75acce..13b171217a4 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/zookeeper/LocalBookkeeperEnsemble.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/zookeeper/LocalBookkeeperEnsemble.java
@@ -297,22 +297,16 @@ public class LocalBookkeeperEnsemble {
}
int bookiePort = portManager.get();
-
+ String bookieId = "bk" + i + "test";
// Ensure registration Z-nodes are cleared when standalone service is restarted ungracefully
- String registrationZnode = String.format("/ledgers/available/%s:%d",
- baseConf.getAdvertisedAddress(), bookiePort);
- if (zkc.exists(registrationZnode, null) != null) {
- try {
- zkc.delete(registrationZnode, -1);
- } catch (NoNodeException nne) {
- // Ignore if z-node was just expired
- }
- }
+ deleteBookieRegistrationZnode(
+ String.format("/ledgers/available/%s:%d", baseConf.getAdvertisedAddress(), bookiePort));
+ deleteBookieRegistrationZnode(String.format("/ledgers/available/%s", bookieId));
bsConfs[i] = new ServerConfiguration(baseConf);
// override settings
bsConfs[i].setBookiePort(bookiePort);
- bsConfs[i].setBookieId("bk" + i + "test");
+ bsConfs[i].setBookieId(bookieId);
String zkServers = "127.0.0.1:" + zkPort;
String metadataServiceUriStr = "zk://" + zkServers + "/ledgers";
@@ -327,6 +321,16 @@ public class LocalBookkeeperEnsemble {
}
}
+ private void deleteBookieRegistrationZnode(String registrationZnode) throws InterruptedException, KeeperException {
+ if (zkc.exists(registrationZnode, null) != null) {
+ try {
+ zkc.delete(registrationZnode, -1);
+ } catch (NoNodeException nne) {
+ // Ignore if z-node was just expired
+ }
+ }
+ }
+
public void runStreamStorage(CompositeConfiguration conf) throws Exception {
String zkServers = "127.0.0.1:" + zkPort;
String metadataServiceUriStr = "zk://" + zkServers + "/ledgers";
(pulsar) 01/02: [fix][client] Fix Reader.hasMessageAvailable return wrong value after seeking by timestamp with startMessageIdInclusive (#23502)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit b7b69dcf817eca708bd71a94521476e9110ac778
Author: Jiawen Wang <74...@users.noreply.github.com>
AuthorDate: Thu Oct 24 17:34:31 2024 +0800
[fix][client] Fix Reader.hasMessageAvailable return wrong value after seeking by timestamp with startMessageIdInclusive (#23502)
(cherry picked from commit fcb359299e9fd784afcebbc3967927a7f1f7c25c)
---
.../org/apache/pulsar/client/impl/ReaderTest.java | 40 ++++++++++++++++++++++
.../apache/pulsar/client/impl/ConsumerImpl.java | 2 ++
2 files changed, 42 insertions(+)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ReaderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ReaderTest.java
index df9a97c29eb..5307d07e9f4 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ReaderTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ReaderTest.java
@@ -855,6 +855,46 @@ public class ReaderTest extends MockedPulsarServiceBaseTest {
}
}
+ @Test
+ public void testHasMessageAvailableAfterSeekTimestampWithMessageIdInclusive() throws Exception {
+ final String topic = "persistent://my-property/my-ns/" +
+ "testHasMessageAvailableAfterSeekTimestampWithMessageInclusive";
+
+ @Cleanup
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topic).create();
+ final long timestampBeforeSend = System.currentTimeMillis();
+ final MessageId sentMsgId = producer.send("msg");
+
+ final List<MessageId> messageIds = new ArrayList<>();
+ messageIds.add(MessageId.earliest);
+ messageIds.add(sentMsgId);
+ messageIds.add(MessageId.latest);
+
+ for (MessageId messageId : messageIds) {
+ @Cleanup
+ Reader<String> reader = pulsarClient.newReader(Schema.STRING).topic(topic).receiverQueueSize(1)
+ .startMessageIdInclusive()
+ .startMessageId(messageId).create();
+ assertTrue(reader.hasMessageAvailable());
+
+ reader.seek(System.currentTimeMillis());
+ assertFalse(reader.hasMessageAvailable());
+ Message<String> message = reader.readNext(10, TimeUnit.SECONDS);
+ assertNull(message);
+ }
+
+ for (MessageId messageId : messageIds) {
+ @Cleanup
+ Reader<String> reader = pulsarClient.newReader(Schema.STRING).topic(topic).receiverQueueSize(1)
+ .startMessageIdInclusive()
+ .startMessageId(messageId).create();
+ assertTrue(reader.hasMessageAvailable());
+
+ reader.seek(timestampBeforeSend);
+ assertTrue(reader.hasMessageAvailable());
+ }
+ }
+
@Test
public void testReaderBuilderStateOnRetryFailure() throws Exception {
String ns = "my-property/my-ns";
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
index 022b194072d..1d5ed1ec400 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
@@ -2426,6 +2426,8 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
.result();
if (lastMessageId.getEntryId() < 0) {
completehasMessageAvailableWithValue(booleanFuture, false);
+ } else if (hasSoughtByTimestamp) {
+ completehasMessageAvailableWithValue(booleanFuture, result < 0);
} else {
completehasMessageAvailableWithValue(booleanFuture,
resetIncludeHead ? result <= 0 : result < 0);