You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by pe...@apache.org on 2021/06/29 03:59:50 UTC
[pulsar] branch branch-2.7 updated (7415036 -> 3d47633)
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a change to branch branch-2.7
in repository https://gitbox.apache.org/repos/asf/pulsar.git.
from 7415036 Change test group to broker for ReplicatorTest and fix the test (#11134)
new 0622649 Print message metadata when getting message by id (#11092)
new 3d47633 Fix replay topic policy message not work (#11136)
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:
.../broker/admin/impl/PersistentTopicsBase.java | 2 ++
.../SystemTopicBasedTopicPoliciesService.java | 10 ++++++++-
.../pulsar/broker/admin/TopicPoliciesTest.java | 25 ++++++++++++++++++++--
.../pulsar/client/admin/internal/TopicsImpl.java | 21 ++++++++++++------
.../org/apache/pulsar/admin/cli/CmdTopics.java | 11 ++++++++++
5 files changed, 59 insertions(+), 10 deletions(-)
[pulsar] 01/02: Print message metadata when getting message by id
(#11092)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.7
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 0622649b2cdb3d0b8fac943972e025e70f5a3200
Author: lipenghui <pe...@apache.org>
AuthorDate: Sat Jun 26 00:27:00 2021 +0800
Print message metadata when getting message by id (#11092)
```
Batch Message ID: 10:1:0
Properties:
"X-Pulsar-batch-size 26825"
"X-Pulsar-num-batch-message 26"
"publish-time 2021-06-25T16:00:40.919+08:00"
```
(cherry picked from commit cadf59d18d6f71f1da06f425709bffa1b10fcf11)
---
.../broker/admin/impl/PersistentTopicsBase.java | 2 ++
.../pulsar/client/admin/internal/TopicsImpl.java | 21 ++++++++++++++-------
.../java/org/apache/pulsar/admin/cli/CmdTopics.java | 11 +++++++++++
3 files changed, 27 insertions(+), 7 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
index 96377de..541a2b9 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
@@ -2336,6 +2336,8 @@ public class PersistentTopicsBase extends AdminResource {
}
if (metadata.hasNumMessagesInBatch()) {
responseBuilder.header("X-Pulsar-num-batch-message", metadata.getNumMessagesInBatch());
+ responseBuilder.header("X-Pulsar-batch-size", metadataAndPayload.readableBytes()
+ - metadata.getSerializedSize());
}
if (metadata.hasNullValue()) {
responseBuilder.header("X-Pulsar-null-value", metadata.hasNullValue());
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java
index 550b7d4..9f2a911 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java
@@ -91,9 +91,10 @@ public class TopicsImpl extends BaseResource implements Topics {
private final WebTarget adminTopics;
private final WebTarget adminV2Topics;
// CHECKSTYLE.OFF: MemberName
- private final String BATCH_HEADER = "X-Pulsar-num-batch-message";
- private final String MESSAGE_ID = "X-Pulsar-Message-ID";
- private final String PUBLISH_TIME = "X-Pulsar-publish-time";
+ private static final String BATCH_HEADER = "X-Pulsar-num-batch-message";
+ private static final String BATCH_SIZE_HEADER = "X-Pulsar-batch-size";
+ private static final String MESSAGE_ID = "X-Pulsar-Message-ID";
+ private static final String PUBLISH_TIME = "X-Pulsar-publish-time";
// CHECKSTYLE.ON: MemberName
public TopicsImpl(WebTarget web, Authentication auth, long readTimeoutMs) {
@@ -1437,11 +1438,11 @@ public class TopicsImpl extends BaseResource implements Topics {
messageMetadata.setNullValue(Boolean.parseBoolean(tmp.toString()));
}
- tmp = headers.getFirst(BATCH_HEADER);
- if (response.getHeaderString(BATCH_HEADER) != null) {
- properties.put(BATCH_HEADER, (String) tmp);
- return getIndividualMsgsFromBatch(topic, msgId, data, properties, messageMetadata);
+ tmp = headers.getFirst(BATCH_SIZE_HEADER);
+ if (tmp != null) {
+ properties.put(BATCH_SIZE_HEADER, (String) tmp);
}
+
for (Entry<String, List<Object>> entry : headers.entrySet()) {
String header = entry.getKey();
if (header.contains("X-Pulsar-PROPERTY-")) {
@@ -1450,6 +1451,12 @@ public class TopicsImpl extends BaseResource implements Topics {
}
}
+ tmp = headers.getFirst(BATCH_HEADER);
+ if (response.getHeaderString(BATCH_HEADER) != null) {
+ properties.put(BATCH_HEADER, (String) tmp);
+ return getIndividualMsgsFromBatch(topic, msgId, data, properties, messageMetadata);
+ }
+
return Collections.singletonList(new MessageImpl<byte[]>(topic, msgId, properties,
Unpooled.wrappedBuffer(data), Schema.BYTES, messageMetadata));
}
diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java
index 65679d1..dbc9ff2 100644
--- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java
+++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java
@@ -884,6 +884,17 @@ public class CmdTopics extends CmdBase {
System.out.println("Cannot find any messages based on ledgerId:"
+ ledgerId + " entryId:" + entryId);
} else {
+ if (message.getMessageId() instanceof BatchMessageIdImpl) {
+ BatchMessageIdImpl msgId = (BatchMessageIdImpl) message.getMessageId();
+ System.out.println("Batch Message ID: " + msgId.getLedgerId() + ":" + msgId.getEntryId() + ":" + msgId.getBatchIndex());
+ } else {
+ MessageIdImpl msgId = (MessageIdImpl) message.getMessageId();
+ System.out.println("Message ID: " + msgId.getLedgerId() + ":" + msgId.getEntryId());
+ }
+ if (message.getProperties().size() > 0) {
+ System.out.println("Properties:");
+ print(message.getProperties());
+ }
ByteBuf date = Unpooled.wrappedBuffer(message.getData());
System.out.println(ByteBufUtil.prettyHexDump(date));
}
[pulsar] 02/02: Fix replay topic policy message not work (#11136)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.7
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 3d476336327720237a28bb1f3d84da39a08b09cd
Author: hangc0276 <ch...@apache.org>
AuthorDate: Tue Jun 29 08:51:06 2021 +0800
Fix replay topic policy message not work (#11136)
### Motivation
When set topic level retention policy for a topic and restart the broker, the topic level retention policy doesn't work.
The reason is when replay the __change_events topic message on `initPolicesCache` stage, it create a reader and read message from earliest and notify the message to update policy for each topic. On update topic policy, it will call getTopicPolicies method.
```Java
public TopicPolicies getTopicPolicies(TopicName topicName) throws TopicPoliciesCacheNotInitException {
if (policyCacheInitMap.containsKey(topicName.getNamespaceObject())
&& !policyCacheInitMap.get(topicName.getNamespaceObject())) {
throw new TopicPoliciesCacheNotInitException();
}
return policiesCache.get(topicName);
}
```
This method will check `policyCacheInitMap` whether init or not for specific namespace.
However, before replay all message completely, the `policyCacheInitMap` keep in not init stage.
Thus the `getTopicPolicies` will throw `TopicPoliciesCacheNotInitException` and the topic policy message will replay failed.
### Modification
1. replay all policy messages after `policyCacheInitMap` initiated.
2. add retention policy check test for broker restart check.
(cherry picked from commit e263e09b1fd6fbe99617edb29b966bed3f5d7c0e)
---
.../SystemTopicBasedTopicPoliciesService.java | 10 ++++++++-
.../pulsar/broker/admin/TopicPoliciesTest.java | 25 ++++++++++++++++++++--
2 files changed, 32 insertions(+), 3 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java
index fb72f3b..29d004d 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java
@@ -251,7 +251,6 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
readerCaches.remove(reader.getSystemTopic().getTopicName().getNamespaceObject());
}
refreshTopicPoliciesCache(msg);
- notifyListener(msg);
if (log.isDebugEnabled()) {
log.debug("[{}] Loop next event reading for system topic.",
reader.getSystemTopic().getTopicName().getNamespaceObject());
@@ -264,6 +263,15 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
}
policyCacheInitMap.computeIfPresent(
reader.getSystemTopic().getTopicName().getNamespaceObject(), (k, v) -> true);
+
+ // replay policy message
+ policiesCache.forEach(((topicName, topicPolicies) -> {
+ if (listeners.get(topicName) != null) {
+ for (TopicPolicyListener<TopicPolicies> listener : listeners.get(topicName)) {
+ listener.onUpdate(topicPolicies);
+ }
+ }
+ }));
future.complete(null);
}
});
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java
index 5f98795..15cd6bc 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java
@@ -615,6 +615,7 @@ public class TopicPoliciesTest extends MockedPulsarServiceBaseTest {
Awaitility.await().atMost(3, TimeUnit.SECONDS)
.until(() -> pulsar.getTopicPoliciesService().cacheIsInitialized(TopicName.get(topic)));
+ // set namespace level inactive topic policies
InactiveTopicPolicies inactiveTopicPolicies =
new InactiveTopicPolicies(InactiveTopicDeleteMode.delete_when_subscriptions_caught_up,100,true);
admin.namespaces().setInactiveTopicPolicies(myNamespace, inactiveTopicPolicies);
@@ -623,13 +624,28 @@ public class TopicPoliciesTest extends MockedPulsarServiceBaseTest {
.untilAsserted(() -> Assert.assertEquals(admin.namespaces().getInactiveTopicPolicies(myNamespace).getInactiveTopicDeleteMode(),
InactiveTopicDeleteMode.delete_when_subscriptions_caught_up));
+ // set namespace retention policies
+ final RetentionPolicies retentionPolicies = new RetentionPolicies(10, -1);
+ admin.namespaces().setRetention(myNamespace, retentionPolicies);
+ Awaitility.await()
+ .untilAsserted(() -> Assert.assertEquals(admin.namespaces().getRetention(myNamespace),
+ retentionPolicies));
+
+ // set topic level inactive topic policies
inactiveTopicPolicies =
new InactiveTopicPolicies(InactiveTopicDeleteMode.delete_when_no_subscriptions,200,false);
admin.topics().setInactiveTopicPolicies(topic, inactiveTopicPolicies);
InactiveTopicPolicies finalInactiveTopicPolicies = inactiveTopicPolicies;
Awaitility.await()
- .untilAsserted(() -> Assert.assertEquals(admin.topics().getInactiveTopicPolicies(topic), finalInactiveTopicPolicies));
+ .untilAsserted(() -> Assert.assertEquals(admin.topics().getInactiveTopicPolicies(topic),
+ finalInactiveTopicPolicies));
+
+ // set topic level retention policies
+ final RetentionPolicies finalRetentionPolicies = new RetentionPolicies(20, -1);
+ admin.topics().setRetention(topic, finalRetentionPolicies);
+ Awaitility.await()
+ .untilAsserted(() -> Assert.assertEquals(admin.topics().getRetention(topic), finalRetentionPolicies));
// restart broker, policy should still take effect
restartBroker();
@@ -637,11 +653,16 @@ public class TopicPoliciesTest extends MockedPulsarServiceBaseTest {
// Trigger the cache init.
Producer<byte[]> producer = pulsarClient.newProducer().topic(topic).create();
-
+ // check inactive topic policies and retention policies.
Awaitility.await()
.untilAsserted(() -> {
PersistentTopic persistentTopic = (PersistentTopic) pulsar.getBrokerService().getTopicIfExists(topic).get().get();
+ ManagedLedgerConfig managedLedgerConfig = persistentTopic.getManagedLedger().getConfig();
Assert.assertEquals(persistentTopic.getInactiveTopicPolicies(), finalInactiveTopicPolicies);
+ Assert.assertEquals(managedLedgerConfig.getRetentionSizeInMB(),
+ finalRetentionPolicies.getRetentionSizeInMB());
+ Assert.assertEquals(managedLedgerConfig.getRetentionTimeMillis(),
+ TimeUnit.MINUTES.toMillis(finalRetentionPolicies.getRetentionTimeInMinutes()));
});
producer.close();