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/26 12:16:27 UTC
[pulsar] branch branch-2.7 updated (531da2c -> 98045aa)
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 531da2c [Broker] Fix direct memory leak in getLastMessageId (#10977)
new 2f9b672 [Broker] Fix the backlog issue with --precise-backlog=true (#10966)
new 81cbc96 Fix hasMessageAvailable return true but can't read message (#10414)
new 78ff879 Fix authorization error if partition number of partitioned topic is updated. (#10300) (#10333)
new 98045aa Use https for the confluent repository
The 4 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:
.../bookkeeper/mledger/impl/ManagedCursorImpl.java | 4 +-
.../bookkeeper/mledger/impl/ManagedCursorTest.java | 30 +++++++
pom.xml | 2 +-
.../authorization/PulsarAuthorizationProvider.java | 21 ++++-
.../apache/pulsar/client/impl/ConsumerImpl.java | 3 +
.../server/ProxyWithJwtAuthorizationTest.java | 91 ++++++++++++++++++++++
6 files changed, 146 insertions(+), 5 deletions(-)
[pulsar] 02/04: Fix hasMessageAvailable return true but can't read
message (#10414)
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 81cbc960055c1480072308b776ce15ac055c237c
Author: feynmanlin <fe...@tencent.com>
AuthorDate: Tue May 4 10:39:35 2021 +0800
Fix hasMessageAvailable return true but can't read message (#10414)
### Motivation
I temporarily fixed this problem in PR https://github.com/apache/pulsar/pull/10190.
Now we have found a better way, this way can avoid the seek, then avoid trigger another reconnection.
Thank you @codelipenghui to troubleshoot this issue with me all night.
We have added a lot of log and found that this issue is caused by some race condition problems. Here is the first reason:
https://github.com/apache/pulsar/blob/f2d72c9fc13a33df584ec1bd96a4c147774b858d/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java#L1808-L1818
Now we have an acknowledgmentsGroupingTracker to filter duplicate messages, and this Tracker will be cleaned up after seek.
However, it is possible that the connection is ready and Broker has pushed message, but `acknowledgmentsGroupingTracker.flushAndClean(); ` has not been executed yet.
Finally hasMessageAvailableAsync returns true, but the message cannot be read because it is filtered by the acknowledgmentsGroupingTracker
### Modifications
clean the tracker when connection was open
### Verifying this change
(cherry picked from commit f69a03b28d29689eaf910e0b99994fa250adb213)
---
.../src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java | 3 +++
1 file changed, 3 insertions(+)
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 2eefb5b..479ddce 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
@@ -839,6 +839,9 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
log.info("[{}][{}] Subscribing to topic on cnx {}, consumerId {}", topic, subscription, cnx.ctx().channel(),consumerId);
long requestId = client.newRequestId();
+ if (duringSeek.get()) {
+ acknowledgmentsGroupingTracker.flushAndClean();
+ }
int currentSize;
synchronized (this) {
[pulsar] 03/04: Fix authorization error if partition number of
partitioned topic is updated. (#10300) (#10333)
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 78ff8792b975f6a1314a2d330e1e150ce5906133
Author: Shen Liu <li...@126.com>
AuthorDate: Thu May 6 08:31:08 2021 +0800
Fix authorization error if partition number of partitioned topic is updated. (#10300) (#10333)
Fixes #10300
### Motivation
Fix the bug that after updating the partition number of a partitioned topic, which has topic level auth policy, new producer/consumer of this topic will get error.
### Modifications
In [`org.apache.pulsar.broker.authorization.PulsarAuthorizationProvider#checkPermission`](https://github.com/apache/pulsar/blob/889b9b8e5efc62d2d0cbc761205fba5759c97af0/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java#L394), if current `topicName` is a sub partition topic, also check the permissions of its partitioned topic.
(cherry picked from commit c1d516254f656e029d6daf6f4ca8914d20ff0d1b)
---
.../authorization/PulsarAuthorizationProvider.java | 21 ++++-
.../server/ProxyWithJwtAuthorizationTest.java | 91 ++++++++++++++++++++++
2 files changed, 110 insertions(+), 2 deletions(-)
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
index e20c89c..934d058 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
@@ -449,16 +449,33 @@ public class PulsarAuthorizationProvider implements AuthorizationProvider {
return;
}
}
+
+ // If the partition number of the partitioned topic having topic level policy is updated,
+ // the new sub partitions may not inherit the policy of the partition topic.
+ // We can also check the permission of partitioned topic.
+ // For https://github.com/apache/pulsar/issues/10300
+ if (topicName.isPartitioned()) {
+ topicRoles = policies.get().auth_policies.destination_auth.get(topicName.getPartitionedTopicName());
+ if (topicRoles != null) {
+ // Topic has custom policy
+ Set<AuthAction> topicActions = topicRoles.get(role);
+ if (topicActions != null && topicActions.contains(action)) {
+ // The role has topic level permission
+ permissionFuture.complete(true);
+ return;
+ }
+ }
+ }
}
permissionFuture.complete(false);
}).exceptionally(ex -> {
- log.warn("Client with Role - {} failed to get permissions for topic - {}. {}", role, topicName,
+ log.warn("Client with Role - {} failed to get permissions for topic - {}. {}", role, topicName,
ex.getMessage());
permissionFuture.completeExceptionally(ex);
return null;
});
} catch (Exception e) {
- log.warn("Client with Role - {} failed to get permissions for topic - {}. {}", role, topicName,
+ log.warn("Client with Role - {} failed to get permissions for topic - {}. {}", role, topicName,
e.getMessage());
permissionFuture.completeExceptionally(e);
}
diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithJwtAuthorizationTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithJwtAuthorizationTest.java
index f683adf..4a7d0b8 100644
--- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithJwtAuthorizationTest.java
+++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithJwtAuthorizationTest.java
@@ -203,6 +203,97 @@ public class ProxyWithJwtAuthorizationTest extends ProducerConsumerBase {
/**
* <pre>
+ * 1. Create a 2-partition topic and grant produce/consume permission to client role.
+ * 2. Use producer/consumer with client role to process the topic, which is fine.
+ * 2. Update the topic partition number to 4.
+ * 3. Use new producer/consumer with client role to process the topic.
+ * 4. Broker should authorize producer/consumer normally.
+ * </pre>
+ */
+ @Test
+ public void testUpdatePartitionNumAndReconnect() throws Exception {
+ log.info("-- Starting {} test --", methodName);
+
+ startProxy();
+ createAdminClient();
+ PulsarClient proxyClient = createPulsarClient(proxyService.getServiceUrl(), PulsarClient.builder());
+
+ String clusterName = "proxy-authorization";
+ String namespaceName = "my-property/my-ns";
+ String topicName = "persistent://my-property/my-ns/my-topic1";
+ String subscriptionName = "my-subscriber-name";
+
+ admin.clusters().createCluster(clusterName, new ClusterData(brokerUrl.toString()));
+
+ admin.tenants().createTenant("my-property",
+ new TenantInfo(Sets.newHashSet(), Sets.newHashSet(clusterName)));
+ admin.namespaces().createNamespace(namespaceName);
+ admin.topics().createPartitionedTopic(topicName, 2);
+ admin.topics().grantPermission(topicName, CLIENT_ROLE,
+ Sets.newHashSet(AuthAction.consume, AuthAction.produce));
+
+ Consumer<byte[]> consumer = proxyClient.newConsumer()
+ .topic(topicName)
+ .subscriptionName(subscriptionName).subscribe();
+
+ Producer<byte[]> producer = proxyClient.newProducer(Schema.BYTES)
+ .topic(topicName).create();
+ final int MSG_NUM = 10;
+ Set<String> messageSet = Sets.newHashSet();
+ for (int i = 0; i < MSG_NUM; i++) {
+ String message = "my-message-" + i;
+ messageSet.add(message);
+ producer.send(message.getBytes());
+ }
+
+ Message<byte[]> msg;
+ Set<String> receivedMessageSet = Sets.newHashSet();
+ for (int i = 0; i < MSG_NUM; i++) {
+ msg = consumer.receive(5, TimeUnit.SECONDS);
+ String receivedMessage = new String(msg.getData());
+ log.debug("Received message: [{}]", receivedMessage);
+ String expectedMessage = "my-message-" + i;
+ receivedMessageSet.add(expectedMessage);
+ consumer.acknowledgeAsync(msg);
+ }
+ Assert.assertEquals(messageSet, receivedMessageSet);
+ consumer.close();
+ producer.close();
+
+ // update partition num
+ admin.topics().updatePartitionedTopic(topicName, 4);
+
+ // produce/consume the topic again
+ consumer = proxyClient.newConsumer()
+ .topic(topicName)
+ .subscriptionName(subscriptionName).subscribe();
+ producer = proxyClient.newProducer(Schema.BYTES)
+ .topic(topicName).create();
+
+ messageSet.clear();
+ for (int i = 0; i < MSG_NUM; i++) {
+ String message = "my-message-" + i;
+ messageSet.add(message);
+ producer.send(message.getBytes());
+ }
+
+ receivedMessageSet.clear();
+ for (int i = 0; i < MSG_NUM; i++) {
+ msg = consumer.receive(5, TimeUnit.SECONDS);
+ String receivedMessage = new String(msg.getData());
+ log.debug("Received message: [{}]", receivedMessage);
+ String expectedMessage = "my-message-" + i;
+ receivedMessageSet.add(expectedMessage);
+ consumer.acknowledgeAsync(msg);
+ }
+ Assert.assertEquals(messageSet, receivedMessageSet);
+ consumer.close();
+ producer.close();
+ log.info("-- Exiting {} test --", methodName);
+ }
+
+ /**
+ * <pre>
* It verifies jwt + Authentication + Authorization (client -> proxy -> broker).
* It also test `SubscriptionAuthMode.Prefix` mode.
*
[pulsar] 04/04: Use https for the confluent repository
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 98045aad8abf52185f15a707b6f4ff83eaefcd63
Author: penghui <pe...@apache.org>
AuthorDate: Sat Jun 26 20:15:03 2021 +0800
Use https for the confluent repository
---
pom.xml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/pom.xml b/pom.xml
index 28011b4..711c6d9 100644
--- a/pom.xml
+++ b/pom.xml
@@ -1731,7 +1731,7 @@ flexible messaging model and an intuitive client API.</description>
</repository>
<repository>
<id>confluent</id>
- <url>http://packages.confluent.io/maven/</url>
+ <url>https://packages.confluent.io/maven/</url>
</repository>
</repositories>
[pulsar] 01/04: [Broker] Fix the backlog issue with
--precise-backlog=true (#10966)
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 2f9b6721c579300e6b1db3f0c062e8dbe38d1925
Author: congbo <39...@users.noreply.github.com>
AuthorDate: Tue Jun 22 07:31:53 2021 +0800
[Broker] Fix the backlog issue with --precise-backlog=true (#10966)
fix backlog issuse with --precise-backlog=true.
Now when `managedLedger` create a new `ledger` complete. if `markDelete` is the `previousLedger` LAC it will delete the previousLedger from `managedLedger` . when get backlog we will use range.close to get `getNumberOfEntries` -1, if previousLedger not exist will get the wrong number.
use range.openClose() to `getBacklog`.
Add the tests for it
(cherry picked from commit e3a97ee4caa80522663f5900b07ae331927de68e)
---
.../bookkeeper/mledger/impl/ManagedCursorImpl.java | 4 +--
.../bookkeeper/mledger/impl/ManagedCursorTest.java | 30 ++++++++++++++++++++++
2 files changed, 32 insertions(+), 2 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
index f732a75..8f671c5 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
@@ -858,13 +858,13 @@ public class ManagedCursorImpl implements ManagedCursor {
messagesConsumedCounter, markDeletePosition, readPosition);
}
if (isPrecise) {
- return getNumberOfEntries(Range.closed(markDeletePosition, ledger.getLastPosition())) - 1;
+ return getNumberOfEntries(Range.openClosed(markDeletePosition, ledger.getLastPosition()));
}
long backlog = ManagedLedgerImpl.ENTRIES_ADDED_COUNTER_UPDATER.get(ledger) - messagesConsumedCounter;
if (backlog < 0) {
// In some case the counters get incorrect values, fall back to the precise backlog count
- backlog = getNumberOfEntries(Range.closed(markDeletePosition, ledger.getLastPosition())) - 1;
+ backlog = getNumberOfEntries(Range.openClosed(markDeletePosition, ledger.getLastPosition()));
}
return backlog;
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java
index 4f6e1a0..cfa7f0b 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java
@@ -45,6 +45,8 @@ import java.util.List;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.Callable;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
@@ -79,6 +81,7 @@ import org.apache.bookkeeper.mledger.ManagedLedgerFactoryConfig;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl.VoidCallback;
import org.apache.bookkeeper.mledger.impl.MetaStore.MetaStoreCallback;
+import org.apache.bookkeeper.mledger.proto.MLDataFormats;
import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo;
import org.apache.bookkeeper.mledger.proto.MLDataFormats.PositionInfo;
import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
@@ -3421,5 +3424,32 @@ public class ManagedCursorTest extends MockedBookKeeperTestCase {
dirtyFactory.shutdown();
}
+ @Test
+ public void testCursorGetBacklog() throws Exception {
+ ManagedLedgerConfig managedLedgerConfig = new ManagedLedgerConfig();
+ managedLedgerConfig.setMaxEntriesPerLedger(2);
+ managedLedgerConfig.setMinimumRolloverTime(0, TimeUnit.MILLISECONDS);
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("get-backlog", managedLedgerConfig);
+ ManagedCursor managedCursor = ledger.openCursor("test");
+
+ PositionImpl position = (PositionImpl) ledger.addEntry("test".getBytes(Encoding));
+ ledger.addEntry("test".getBytes(Encoding));
+ PositionImpl position1 = (PositionImpl) ledger.addEntry("test".getBytes(Encoding));
+ ledger.addEntry("test".getBytes(Encoding));
+
+ Assert.assertEquals(managedCursor.getNumberOfEntriesInBacklog(true), 4);
+ Assert.assertEquals(managedCursor.getNumberOfEntriesInBacklog(false), 4);
+ Field field = ManagedLedgerImpl.class.getDeclaredField("ledgers");
+ field.setAccessible(true);
+
+ ((ConcurrentSkipListMap<Long, MLDataFormats.ManagedLedgerInfo.LedgerInfo>) field.get(ledger)).remove(position.ledgerId);
+ field = ManagedCursorImpl.class.getDeclaredField("markDeletePosition");
+ field.setAccessible(true);
+ field.set(managedCursor, PositionImpl.get(position1.ledgerId, -1));
+
+
+ Assert.assertEquals(managedCursor.getNumberOfEntriesInBacklog(true), 2);
+ Assert.assertEquals(managedCursor.getNumberOfEntriesInBacklog(false), 4);
+ }
private static final Logger log = LoggerFactory.getLogger(ManagedCursorTest.class);
}