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 2022/01/10 16:07:48 UTC
[pulsar] branch branch-2.9 updated (2547d5c -> a08f0f5)
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a change to branch branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git.
from 2547d5c Upgrade Gson version 2.8.6 to 2.8.9 (#13610)
new 406aa4e Fix the null point caused by deleting the system topic policy (#12367)
new 395bd73 fix(Auth): Fix multi roles authz cannot handle empty roles case (#13477)
new 7df7896 Change ContextClassLoader to NarClassLoader in AdditionalServlet (#13501)
new 25fece0 Bump slf4j to 1.7.32
new f1f2f3d [Presto] Fix issue of mixed SLF4J library versions in presto libs (#13603)
new 624bf12 [Tests] Fix flakiness issue when spying ServerCnx (#13608)
new 9eef658 update log content (#13540)
new 6fd7421 Fix NPE when unloading namespace bundle (#13571)
new 598cf40 [Tests] Upgrade Mockito to latest stable 3.x version, 3.12.4 (#13622)
new a27fb57 Fix time field use error (#12249)
new 62b4c60 Fix issues 11964, deadlock bug when use key_shared mode (#11965)
new 89c7b6e [issue #13439][backlogQuota] Allow both limit and limitsize be null (#13557)
new 23700db Fix reader skipped remaining compacted data during the topic unloading. (#13629)
new d090413 Update notice year. (#13653)
new ad043fc Remove txn add partition info log (#13670)
new a08f0f5 [Java Client] fixed Producer semaphore permit release issue (#13682)
The 16 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:
NOTICE | 2 +-
buildtools/pom.xml | 2 +-
distribution/server/src/assemble/LICENSE.bin.txt | 6 +-
distribution/server/src/assemble/NOTICE.bin.txt | 2 +-
.../apache/bookkeeper/mledger/ManagedLedger.java | 3 +-
.../bookkeeper/mledger/impl/ManagedCursorImpl.java | 4 +-
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 7 +-
.../mledger/impl/NonDurableCursorImpl.java | 23 +++--
pom.xml | 4 +-
.../apache/pulsar/broker/ClassLoaderSwitcher.java | 22 ++--
.../MultiRolesTokenAuthorizationProvider.java | 3 +
.../servlet/AdditionalServletWithClassLoader.java | 18 +++-
.../MultiRolesTokenAuthorizationProviderTest.java | 32 +++++-
.../loadbalance/impl/ModularLoadManagerImpl.java | 3 +-
.../protocol/ProtocolHandlerWithClassLoader.java | 19 +---
.../AbstractDispatcherSingleActiveConsumer.java | 4 -
.../pulsar/broker/service/BrokerService.java | 18 ++--
.../apache/pulsar/broker/service/ServerCnx.java | 1 -
.../SystemTopicBasedTopicPoliciesService.java | 27 +++--
.../broker/service/persistent/PersistentTopic.java | 8 +-
.../pulsar/broker/admin/TopicPoliciesTest.java | 23 +++++
.../PersistentDispatcherFailoverConsumerTest.java | 10 +-
.../service/PersistentTopicConcurrentTest.java | 6 +-
.../pulsar/broker/service/PersistentTopicTest.java | 14 ++-
.../AdditionalServletWithClassLoaderTest.java | 114 +++++++++++++++++++++
.../pulsar/client/impl/ProducerSemaphoreTest.java | 14 +++
.../pulsar/compaction/CompactedTopicTest.java | 55 ++++++++++
.../policies/data/impl/BacklogQuotaImpl.java | 4 +-
.../apache/pulsar/client/impl/ProducerImpl.java | 16 ++-
.../metadata/BacklogQuotaCompatibilityTest.java | 20 ++++
pulsar-sql/pom.xml | 12 +++
pulsar-sql/presto-distribution/LICENSE | 12 +--
.../pulsar/sql/presto/PulsarRecordCursor.java | 2 +-
.../offload/jcloud/impl/MockManagedLedger.java | 4 +-
34 files changed, 413 insertions(+), 101 deletions(-)
copy pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/CommaWindowFunction.java => pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ClassLoaderSwitcher.java (59%)
create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/web/plugin/servlet/AdditionalServletWithClassLoaderTest.java
[pulsar] 03/16: Change ContextClassLoader to NarClassLoader in AdditionalServlet (#13501)
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.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 7df7896ec5bb52d3890bde161375f68a81d830c5
Author: Qiang Zhao <74...@users.noreply.github.com>
AuthorDate: Tue Jan 4 22:34:35 2022 +0800
Change ContextClassLoader to NarClassLoader in AdditionalServlet (#13501)
### Motivation
It's ``AddtionalServlet`` side change, like #11270
### Modifications
Change context class loader through Thread.currentThread().setContextClassLoader(classLoader) before every protocol handler callback, and change it back to original class loader afterwards.
(cherry picked from commit 40ac7934ed567b2b9f16dbd7f937f7973dcddc21)
---
.../apache/pulsar/broker/ClassLoaderSwitcher.java | 37 +++++++
.../servlet/AdditionalServletWithClassLoader.java | 18 +++-
.../protocol/ProtocolHandlerWithClassLoader.java | 19 +---
.../AdditionalServletWithClassLoaderTest.java | 114 +++++++++++++++++++++
4 files changed, 166 insertions(+), 22 deletions(-)
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ClassLoaderSwitcher.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ClassLoaderSwitcher.java
new file mode 100644
index 0000000..787182e
--- /dev/null
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ClassLoaderSwitcher.java
@@ -0,0 +1,37 @@
+/**
+ * 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;
+
+/**
+ * Help to switch the class loader of current thread to the NarClassLoader, and change it back when it's done.
+ * With the help of try-with-resources statement, the code would be cleaner than using try finally every time.
+ */
+public class ClassLoaderSwitcher implements AutoCloseable {
+ private final ClassLoader prevClassLoader;
+
+ public ClassLoaderSwitcher(ClassLoader classLoader) {
+ prevClassLoader = Thread.currentThread().getContextClassLoader();
+ Thread.currentThread().setContextClassLoader(classLoader);
+ }
+
+ @Override
+ public void close() {
+ Thread.currentThread().setContextClassLoader(prevClassLoader);
+ }
+}
\ No newline at end of file
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/plugin/servlet/AdditionalServletWithClassLoader.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/plugin/servlet/AdditionalServletWithClassLoader.java
index 06a51c8..7f1af73 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/plugin/servlet/AdditionalServletWithClassLoader.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/web/plugin/servlet/AdditionalServletWithClassLoader.java
@@ -19,9 +19,11 @@
package org.apache.pulsar.broker.web.plugin.servlet;
import java.io.IOException;
+
import lombok.Data;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.broker.ClassLoaderSwitcher;
import org.apache.pulsar.common.configuration.PulsarConfiguration;
import org.apache.pulsar.common.nar.NarClassLoader;
import org.eclipse.jetty.servlet.ServletHolder;
@@ -39,22 +41,30 @@ public class AdditionalServletWithClassLoader implements AdditionalServlet {
@Override
public void loadConfig(PulsarConfiguration pulsarConfiguration) {
- servlet.loadConfig(pulsarConfiguration);
+ try (ClassLoaderSwitcher ignored = new ClassLoaderSwitcher(classLoader)) {
+ servlet.loadConfig(pulsarConfiguration);
+ }
}
@Override
public String getBasePath() {
- return servlet.getBasePath();
+ try (ClassLoaderSwitcher ignored = new ClassLoaderSwitcher(classLoader)) {
+ return servlet.getBasePath();
+ }
}
@Override
public ServletHolder getServletHolder() {
- return servlet.getServletHolder();
+ try (ClassLoaderSwitcher ignored = new ClassLoaderSwitcher(classLoader)) {
+ return servlet.getServletHolder();
+ }
}
@Override
public void close() {
- servlet.close();
+ try (ClassLoaderSwitcher ignored = new ClassLoaderSwitcher(classLoader)) {
+ servlet.close();
+ }
try {
classLoader.close();
} catch (IOException e) {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/protocol/ProtocolHandlerWithClassLoader.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/protocol/ProtocolHandlerWithClassLoader.java
index 223cf81..63aa669 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/protocol/ProtocolHandlerWithClassLoader.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/protocol/ProtocolHandlerWithClassLoader.java
@@ -26,6 +26,7 @@ import java.util.Map;
import lombok.Data;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.broker.ClassLoaderSwitcher;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.service.BrokerService;
import org.apache.pulsar.common.nar.NarClassLoader;
@@ -95,22 +96,4 @@ class ProtocolHandlerWithClassLoader implements ProtocolHandler {
log.warn("Failed to close the protocol handler class loader", e);
}
}
-
- /**
- * Help to switch the class loader of current thread to the NarClassLoader, and change it back when it's done.
- * With the help of try-with-resources statement, the code would be cleaner than using try finally every time.
- */
- private static class ClassLoaderSwitcher implements AutoCloseable {
- private final ClassLoader prevClassLoader;
-
- ClassLoaderSwitcher(ClassLoader classLoader) {
- prevClassLoader = Thread.currentThread().getContextClassLoader();
- Thread.currentThread().setContextClassLoader(classLoader);
- }
-
- @Override
- public void close() {
- Thread.currentThread().setContextClassLoader(prevClassLoader);
- }
- }
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/plugin/servlet/AdditionalServletWithClassLoaderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/plugin/servlet/AdditionalServletWithClassLoaderTest.java
new file mode 100644
index 0000000..c875e8a
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/plugin/servlet/AdditionalServletWithClassLoaderTest.java
@@ -0,0 +1,114 @@
+/**
+ * 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.web.plugin.servlet;
+
+import static org.mockito.ArgumentMatchers.same;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNull;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.common.configuration.PulsarConfiguration;
+import org.apache.pulsar.common.nar.NarClassLoader;
+import org.eclipse.jetty.servlet.ServletHolder;
+import org.testng.annotations.Test;
+
+
+/**
+ * Unit test {@link AdditionalServletWithClassLoader}.
+ */
+@Test(groups = "broker")
+public class AdditionalServletWithClassLoaderTest {
+
+ @Test
+ public void testWrapper() {
+ AdditionalServlet servlet = mock(AdditionalServlet.class);
+ NarClassLoader loader = mock(NarClassLoader.class);
+ AdditionalServletWithClassLoader wrapper = new AdditionalServletWithClassLoader(servlet, loader);
+ // test getBasePath
+ String basePath = "bathPath";
+ when(servlet.getBasePath()).thenReturn(basePath);
+ assertEquals(basePath, wrapper.getBasePath());
+ verify(servlet, times(1)).getBasePath();
+ // test loadConfig
+ ServiceConfiguration conf = new ServiceConfiguration();
+ wrapper.loadConfig(conf);
+ verify(servlet, times(1)).loadConfig(same(conf));
+ // test getServlet
+ assertEquals(wrapper.getServlet(),servlet);
+ // test getServletHolder
+ ServletHolder servletHolder = new ServletHolder();
+ when(servlet.getServletHolder()).thenReturn(servletHolder);
+ assertEquals(wrapper.getServletHolder(),servletHolder);
+ verify(servlet, times(1)).getServletHolder();
+ }
+
+ @Test
+ public void testClassLoaderSwitcher() throws Exception {
+ NarClassLoader narLoader = mock(NarClassLoader.class);
+ AdditionalServlet servlet = new AdditionalServlet() {
+ @Override
+ public void loadConfig(PulsarConfiguration pulsarConfiguration) {
+ assertEquals(Thread.currentThread().getContextClassLoader(), narLoader);
+ }
+
+ @Override
+ public String getBasePath() {
+ assertEquals(Thread.currentThread().getContextClassLoader(), narLoader);
+ return "base-path";
+ }
+
+ @Override
+ public ServletHolder getServletHolder() {
+ assertEquals(Thread.currentThread().getContextClassLoader(), narLoader);
+ return null;
+ }
+
+ @Override
+ public void close() {
+ assertEquals(Thread.currentThread().getContextClassLoader(), narLoader);
+ }
+ };
+
+ AdditionalServletWithClassLoader additionalServletWithClassLoader =
+ new AdditionalServletWithClassLoader(servlet, narLoader);
+ ClassLoader curClassLoader = Thread.currentThread().getContextClassLoader();
+ // test class loader
+ assertEquals(additionalServletWithClassLoader.getClassLoader(), narLoader);
+ // test getBasePath
+ assertEquals(additionalServletWithClassLoader.getBasePath(), "base-path");
+ assertEquals(Thread.currentThread().getContextClassLoader(), curClassLoader);
+ // test loadConfig
+ ServiceConfiguration conf = new ServiceConfiguration();
+ additionalServletWithClassLoader.loadConfig(conf);
+ assertEquals(Thread.currentThread().getContextClassLoader(), curClassLoader);
+ // test getServletHolder
+ assertNull(additionalServletWithClassLoader.getServletHolder());
+ assertEquals(Thread.currentThread().getContextClassLoader(), curClassLoader);
+ // test getServlet
+ assertEquals(additionalServletWithClassLoader.getServlet(), servlet);
+ assertEquals(Thread.currentThread().getContextClassLoader(), curClassLoader);
+ // test close
+ additionalServletWithClassLoader.close();
+ assertEquals(Thread.currentThread().getContextClassLoader(), curClassLoader);
+
+ }
+}
[pulsar] 09/16: [Tests] Upgrade Mockito to latest stable 3.x version, 3.12.4 (#13622)
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.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 598cf4099be6fbba25a00ab645adc9e0b67bc10e
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Wed Jan 5 14:54:27 2022 +0200
[Tests] Upgrade Mockito to latest stable 3.x version, 3.12.4 (#13622)
(cherry picked from commit 1f102814b96db723592b2fdfb2bbf288d9b75fb9)
---
pom.xml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/pom.xml b/pom.xml
index 2358fc7..88d2e58 100644
--- a/pom.xml
+++ b/pom.xml
@@ -213,7 +213,7 @@ flexible messaging model and an intuitive client API.</description>
<kerby.version>1.1.1</kerby.version>
<testng.version>7.3.0</testng.version>
<junit4.version>4.13.1</junit4.version>
- <mockito.version>3.8.0</mockito.version>
+ <mockito.version>3.12.4</mockito.version>
<powermock.version>2.0.9</powermock.version>
<javassist.version>3.25.0-GA</javassist.version>
<failsafe.version>2.3.1</failsafe.version>
[pulsar] 15/16: Remove txn add partition info log (#13670)
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.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit ad043fca44450fd8b8b13f27fc3460fc2442fa1a
Author: lipenghui <pe...@apache.org>
AuthorDate: Mon Jan 10 09:44:47 2022 +0800
Remove txn add partition info log (#13670)
During my test, there are many output logs `handle add partition to txn finish` for the broker.
### Documentation
Check the box below or label this PR directly (if you have committer privilege).
(cherry picked from commit 53bd7a44f252812a0ff4c1bc22d0a03e606a4875)
---
.../src/main/java/org/apache/pulsar/broker/service/ServerCnx.java | 1 -
1 file changed, 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 c4ab0e2..583ad3c 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
@@ -2335,7 +2335,6 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
}
ctx.writeAndFlush(Commands.newAddSubscriptionToTxnResponse(requestId,
txnID.getLeastSigBits(), txnID.getMostSigBits()));
- log.info("handle add partition to txn finish.");
} else {
ex = handleTxnException(ex, BaseCommand.Type.ADD_SUBSCRIPTION_TO_TXN.name(), requestId);
[pulsar] 02/16: fix(Auth): Fix multi roles authz cannot handle empty roles case (#13477)
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.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 395bd73b3684602fbdcca40b8e09972632cefb4a
Author: Zike Yang <zk...@streamnative.io>
AuthorDate: Thu Dec 30 19:13:41 2021 +0800
fix(Auth): Fix multi roles authz cannot handle empty roles case (#13477)
Motivation
Currently, if the roles in the token are empty, then he `MultiRolesTokenAuthorizationProvider` will have problems processing it. It will keep waiting for an empty list of futures. Eventually causing the operation to time out.
Modification
* In `MultiRolesTokenAuthorizationProvider.authorize`, return false immediately when the roles are empty.
Signed-off-by: Zike Yang <zk...@streamnative.io>
(cherry picked from commit 4f942d71263424f12d9be88f39b79c41485d0f32)
---
.../MultiRolesTokenAuthorizationProvider.java | 3 ++
.../MultiRolesTokenAuthorizationProviderTest.java | 32 ++++++++++++++++++++--
2 files changed, 33 insertions(+), 2 deletions(-)
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/MultiRolesTokenAuthorizationProvider.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/MultiRolesTokenAuthorizationProvider.java
index dcdf779..8a91d7f 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/MultiRolesTokenAuthorizationProvider.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/MultiRolesTokenAuthorizationProvider.java
@@ -128,6 +128,9 @@ public class MultiRolesTokenAuthorizationProvider extends PulsarAuthorizationPro
public CompletableFuture<Boolean> authorize(AuthenticationDataSource authenticationData, Function<String, CompletableFuture<Boolean>> authorizeFunc) {
List<String> roles = getRoles(authenticationData);
+ if (roles.isEmpty()) {
+ return CompletableFuture.completedFuture(false);
+ }
List<CompletableFuture<Boolean>> futures = new ArrayList<>(roles.size());
roles.forEach(r -> futures.add(authorizeFunc.apply(r)));
return CompletableFuture.supplyAsync(() -> {
diff --git a/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authorization/MultiRolesTokenAuthorizationProviderTest.java b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authorization/MultiRolesTokenAuthorizationProviderTest.java
index fdedf86..edd0baa 100644
--- a/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authorization/MultiRolesTokenAuthorizationProviderTest.java
+++ b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/authorization/MultiRolesTokenAuthorizationProviderTest.java
@@ -56,7 +56,9 @@ public class MultiRolesTokenAuthorizationProviderTest {
};
Assert.assertTrue(provider.authorize(ads, role -> {
- if (role.equals(userB)) return CompletableFuture.completedFuture(true); // only userB has permission
+ if (role.equals(userB)) {
+ return CompletableFuture.completedFuture(true); // only userB has permission
+ }
return CompletableFuture.completedFuture(false);
}).get());
@@ -65,7 +67,33 @@ public class MultiRolesTokenAuthorizationProviderTest {
}).get());
Assert.assertFalse(provider.authorize(ads, role -> {
- return CompletableFuture.completedFuture(false); // only users has no permission
+ return CompletableFuture.completedFuture(false); // all users has no permission
}).get());
}
+
+ @Test
+ public void testMultiRolesAuthzWithEmptyRoles() throws Exception {
+ SecretKey secretKey = AuthTokenUtils.createSecretKey(SignatureAlgorithm.HS256);
+ String token = Jwts.builder().claim("sub", new String[]{}).signWith(secretKey).compact();
+
+ MultiRolesTokenAuthorizationProvider provider = new MultiRolesTokenAuthorizationProvider();
+
+ AuthenticationDataSource ads = new AuthenticationDataSource() {
+ @Override
+ public boolean hasDataFromHttp() {
+ return true;
+ }
+
+ @Override
+ public String getHttpHeader(String name) {
+ if (name.equals("Authorization")) {
+ return "Bearer " + token;
+ } else {
+ throw new IllegalArgumentException("Wrong HTTP header");
+ }
+ }
+ };
+
+ Assert.assertFalse(provider.authorize(ads, role -> CompletableFuture.completedFuture(false)).get());
+ }
}
[pulsar] 11/16: Fix issues 11964, deadlock bug when use key_shared mode (#11965)
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.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 62b4c60b4b01805ee3b374170d20333a8e6bac1d
Author: baomingyu <ba...@163.com>
AuthorDate: Fri Jan 7 12:51:41 2022 +0800
Fix issues 11964, deadlock bug when use key_shared mode (#11965)
(cherry picked from commit fd3ba55f8e39992ecbc96830af5ee58e3247a25a)
---
.../java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java | 4 +++-
1 file changed, 3 insertions(+), 1 deletion(-)
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 3d382ad..2f9578c 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
@@ -1993,7 +1993,6 @@ public class ManagedCursorImpl implements ManagedCursor {
if (individualDeletedMessages.isEmpty()) {
// No changes to individually deleted messages, so nothing to do at this point
- callback.deleteComplete(ctx);
return;
}
@@ -2025,6 +2024,9 @@ public class ManagedCursorImpl implements ManagedCursor {
return;
} finally {
lock.writeLock().unlock();
+ if (individualDeletedMessages.isEmpty()) {
+ callback.deleteComplete(ctx);
+ }
}
// Apply rate limiting to mark-delete operations
[pulsar] 10/16: Fix time field use error (#12249)
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.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit a27fb57a183649eede83da50af5bc9c859535367
Author: liuchangqing <ca...@126.com>
AuthorDate: Thu Jan 6 13:19:24 2022 +0800
Fix time field use error (#12249)
* Fix time field use error
* rebuid
Co-authored-by: liu.changqing <ch...@kingdee.com>
(cherry picked from commit 793dd910fc17b0e149f4b554b54cad55ca5bc479)
---
.../src/main/java/org/apache/pulsar/sql/presto/PulsarRecordCursor.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarRecordCursor.java b/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarRecordCursor.java
index 99be9f9..d839e05 100644
--- a/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarRecordCursor.java
+++ b/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarRecordCursor.java
@@ -580,7 +580,7 @@ public class PulsarRecordCursor implements RecordCursor {
currentRowValuesMap.put(columnHandle, longValueProvider(this.partition));
} else if (PulsarInternalColumn.EVENT_TIME.getName().equals(columnHandle.getName())) {
currentRowValuesMap.put(columnHandle, PulsarFieldValueProviders.timeValueProvider(
- this.currentMessage.getEventTime(), this.currentMessage.getPublishTime() == 0));
+ this.currentMessage.getEventTime(), this.currentMessage.getEventTime() == 0));
} else if (PulsarInternalColumn.PUBLISH_TIME.getName().equals(columnHandle.getName())) {
currentRowValuesMap.put(columnHandle, PulsarFieldValueProviders.timeValueProvider(
this.currentMessage.getPublishTime(), this.currentMessage.getPublishTime() == 0));
[pulsar] 06/16: [Tests] Fix flakiness issue when spying ServerCnx (#13608)
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.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 624bf12934e8e0fccb411321c693b2f601a853b7
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Tue Jan 4 19:32:09 2022 +0200
[Tests] Fix flakiness issue when spying ServerCnx (#13608)
Fixes #13570
(cherry picked from commit de99c2c9c3c54b31096e6d734de8132e96d50c79)
---
.../service/PersistentDispatcherFailoverConsumerTest.java | 10 ++++++++--
.../broker/service/PersistentTopicConcurrentTest.java | 6 +++++-
.../apache/pulsar/broker/service/PersistentTopicTest.java | 14 +++++++++++---
3 files changed, 24 insertions(+), 6 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java
index bc5e05c..84cf739 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java
@@ -21,6 +21,7 @@ package org.apache.pulsar.broker.service;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.matches;
import static org.mockito.ArgumentMatchers.same;
+import static org.mockito.Mockito.CALLS_REAL_METHODS;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
@@ -28,6 +29,7 @@ import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.withSettings;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNull;
import static org.testng.AssertJUnit.assertEquals;
@@ -165,7 +167,9 @@ public class PersistentDispatcherFailoverConsumerTest {
return null;
}).when(channelCtx).writeAndFlush(any(), any());
- serverCnx = spy(new ServerCnx(pulsar));
+ serverCnx = mock(ServerCnx.class, withSettings()
+ .useConstructor(pulsar)
+ .defaultAnswer(CALLS_REAL_METHODS));
doReturn(true).when(serverCnx).isActive();
doReturn(true).when(serverCnx).isWritable();
doReturn(new InetSocketAddress("localhost", 1234)).when(serverCnx).clientAddress();
@@ -174,7 +178,9 @@ public class PersistentDispatcherFailoverConsumerTest {
doReturn(new PulsarCommandSenderImpl(null, serverCnx))
.when(serverCnx).getCommandSender();
- serverCnxWithOldVersion = spy(new ServerCnx(pulsar));
+ serverCnxWithOldVersion = mock(ServerCnx.class, withSettings()
+ .useConstructor(pulsar)
+ .defaultAnswer(CALLS_REAL_METHODS));
doReturn(true).when(serverCnxWithOldVersion).isActive();
doReturn(true).when(serverCnxWithOldVersion).isWritable();
doReturn(new InetSocketAddress("localhost", 1234))
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicConcurrentTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicConcurrentTest.java
index ef81ed2..78c7e25 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicConcurrentTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicConcurrentTest.java
@@ -18,10 +18,12 @@
*/
package org.apache.pulsar.broker.service;
+import static org.mockito.Mockito.CALLS_REAL_METHODS;
import static org.mockito.Mockito.any;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.withSettings;
import static org.testng.Assert.assertFalse;
import java.lang.reflect.Method;
@@ -101,7 +103,9 @@ public class PersistentTopicConcurrentTest extends MockedBookKeeperTestCase {
brokerService = spy(new BrokerService(pulsar, eventLoopGroup));
doReturn(brokerService).when(pulsar).getBrokerService();
- serverCnx = spy(new ServerCnx(pulsar));
+ serverCnx = mock(ServerCnx.class, withSettings()
+ .useConstructor(pulsar)
+ .defaultAnswer(CALLS_REAL_METHODS));
doReturn(true).when(serverCnx).isActive();
NamespaceService nsSvc = mock(NamespaceService.class);
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 f1a11ad..d94320f 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
@@ -22,6 +22,7 @@ import static org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest.createMo
import static org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest.createMockZooKeeper;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.CALLS_REAL_METHODS;
import static org.mockito.Mockito.anyString;
import static org.mockito.Mockito.atLeast;
import static org.mockito.Mockito.doAnswer;
@@ -34,6 +35,7 @@ import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.withSettings;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNull;
@@ -207,7 +209,9 @@ public class PersistentTopicTest extends MockedBookKeeperTestCase {
brokerService = spy(new BrokerService(pulsar, eventLoopGroup));
doReturn(brokerService).when(pulsar).getBrokerService();
- serverCnx = spy(new ServerCnx(pulsar));
+ serverCnx = mock(ServerCnx.class, withSettings()
+ .useConstructor(pulsar)
+ .defaultAnswer(CALLS_REAL_METHODS));
doReturn(true).when(serverCnx).isActive();
doReturn(true).when(serverCnx).isWritable();
doReturn(new InetSocketAddress("localhost", 1234)).when(serverCnx).clientAddress();
@@ -584,7 +588,9 @@ public class PersistentTopicTest extends MockedBookKeeperTestCase {
final String producerNameBase = "producer";
final String role = "appid1";
- ServerCnx cnx = spy(new ServerCnx(pulsar));
+ ServerCnx cnx = mock(ServerCnx.class, withSettings()
+ .useConstructor(pulsar)
+ .defaultAnswer(CALLS_REAL_METHODS));
doReturn(true).when(cnx).isActive();
doReturn(true).when(cnx).isWritable();
doReturn(new InetSocketAddress(address, 1234)).when(cnx).clientAddress();
@@ -1022,7 +1028,9 @@ public class PersistentTopicTest extends MockedBookKeeperTestCase {
final String consumerNameBase = "consumer";
final String role = "appid1";
- ServerCnx cnx = spy(new ServerCnx(pulsar));
+ ServerCnx cnx = mock(ServerCnx.class, withSettings()
+ .useConstructor(pulsar)
+ .defaultAnswer(CALLS_REAL_METHODS));
doReturn(true).when(cnx).isActive();
doReturn(true).when(cnx).isWritable();
doReturn(new InetSocketAddress(address, 1234)).when(cnx).clientAddress();
[pulsar] 01/16: Fix the null point caused by deleting the system topic policy (#12367)
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.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 406aa4e472af612acf5377ad7def4f5e2e64de2f
Author: Zixuan Liu <no...@gmail.com>
AuthorDate: Fri Oct 22 11:03:40 2021 +0800
Fix the null point caused by deleting the system topic policy (#12367)
Signed-off-by: Zixuan Liu <no...@gmail.com>
### Motivation
`Message<PulsarEvent>.getValue()` sometimes returns `null` in `SystemTopicBasedTopicPoliciesService.notifyListener()`, so we need to skip the messages that do not belong to the policy type, this problem can cause the policy service to fail to work.
### Modifications
- Checks the `Message<PulsarEvent>.getValue()` value.
- Uses the `event` instead of `null` as message value when delete policy.
### Verifying this change
- [x] Make sure that the change passes the CI checks.
### Does this pull request potentially affect one of the following parts:
*If `yes` was chosen, please highlight the changes*
- Dependencies (does it add or upgrade a dependency): no
- The public API: no
- The schema: no
- The default values of configurations: no
- The wire protocol: no
- The rest endpoints: no
- The admin cli options: no
- Anything that affects deployment: no
### Documentation
Check the box below and label this PR (if you have committer privilege).
Need to update docs?
- [ ] doc-required
- [x] no-need-doc
- [ ] doc
(cherry picked from commit d310e79f360f41ddabc820fd0e45af67a8a4db82)
---
.../SystemTopicBasedTopicPoliciesService.java | 27 +++++++++++++++-------
.../pulsar/broker/admin/TopicPoliciesTest.java | 23 ++++++++++++++++++
2 files changed, 42 insertions(+), 8 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 bf27736..a6c786b 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
@@ -126,7 +126,7 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
}
}
});
- })
+ })
);
}
});
@@ -149,6 +149,17 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
}
private void notifyListener(Message<PulsarEvent> msg) {
+ // delete policies
+ if (msg.getValue() == null) {
+ TopicName topicName = TopicName.get(TopicName.get(msg.getKey()).getPartitionedTopicName());
+ if (listeners.get(topicName) != null) {
+ for (TopicPolicyListener<TopicPolicies> listener : listeners.get(topicName)) {
+ listener.onUpdate(null);
+ }
+ }
+ return;
+ }
+
if (!EventType.TOPIC_POLICY.equals(msg.getValue().getEventType())) {
return;
}
@@ -296,12 +307,11 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
removeOwnedNamespaceBundleAsync(bundle);
}
- @Override
- public boolean test(NamespaceBundle namespaceBundle) {
- return true;
- }
-
- });
+ @Override
+ public boolean test(NamespaceBundle namespaceBundle) {
+ return true;
+ }
+ });
}
private void initPolicesCache(SystemTopicClient.Reader<PulsarEvent> reader, CompletableFuture<Void> future) {
@@ -449,7 +459,8 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
if (e != null) {
future.completeExceptionally(e);
}
- if (EventType.TOPIC_POLICY.equals(msg.getValue().getEventType())) {
+ if (msg.getValue() != null
+ && EventType.TOPIC_POLICY.equals(msg.getValue().getEventType())) {
TopicPoliciesEvent topicPoliciesEvent = msg.getValue().getTopicPoliciesEvent();
if (topicName.equals(TopicName.get(
topicPoliciesEvent.getDomain(),
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 5047c0c..7af136a 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
@@ -2688,4 +2688,27 @@ public class TopicPoliciesTest extends MockedPulsarServiceBaseTest {
});
}
+ @Test
+ public void testLoopCreateAndDeleteTopicPolicies() throws Exception {
+ final String topic = testTopic + UUID.randomUUID();
+
+ int n = 0;
+ while (n < 2) {
+ n++;
+ pulsarClient.newProducer().topic(topic).create().close();
+ Awaitility.await().untilAsserted(() -> {
+ Assertions.assertThat(pulsar.getTopicPoliciesService().getTopicPolicies(TopicName.get(topic))).isNull();
+ });
+
+ admin.topics().setMaxConsumersPerSubscription(topic, 1);
+ Awaitility.await().untilAsserted(() -> {
+ Assertions.assertThat(pulsar.getTopicPoliciesService().getTopicPolicies(TopicName.get(topic))).isNotNull();
+ });
+
+ admin.topics().delete(topic);
+ Awaitility.await().untilAsserted(() -> {
+ Assertions.assertThat(pulsar.getTopicPoliciesService().getTopicPolicies(TopicName.get(topic))).isNull();
+ });
+ }
+ }
}
[pulsar] 04/16: Bump slf4j to 1.7.32
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.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 25fece02f3b021f8f50ecf171a63225ed6621195
Author: ZhangJian He <sh...@gmail.com>
AuthorDate: Tue Jan 4 20:46:33 2022 +0800
Bump slf4j to 1.7.32
(cherry picked from commit 05ea215ddc40e9957da7ed492d33a11831df3177)
---
buildtools/pom.xml | 2 +-
distribution/server/src/assemble/LICENSE.bin.txt | 6 +++---
pom.xml | 2 +-
pulsar-sql/presto-distribution/LICENSE | 6 +++---
4 files changed, 8 insertions(+), 8 deletions(-)
diff --git a/buildtools/pom.xml b/buildtools/pom.xml
index 207a729..be6f1fe 100644
--- a/buildtools/pom.xml
+++ b/buildtools/pom.xml
@@ -40,7 +40,7 @@
<maven.compiler.target>1.8</maven.compiler.target>
<surefire.version>3.0.0-M3</surefire.version>
<log4j2.version>2.17.1</log4j2.version>
- <slf4j.version>1.7.25</slf4j.version>
+ <slf4j.version>1.7.32</slf4j.version>
<testng.version>7.3.0</testng.version>
<commons-lang3.version>3.11</commons-lang3.version>
<maven-shade-plugin.version>3.2.4</maven-shade-plugin.version>
diff --git a/distribution/server/src/assemble/LICENSE.bin.txt b/distribution/server/src/assemble/LICENSE.bin.txt
index 4e3b9b1..1620c9f 100644
--- a/distribution/server/src/assemble/LICENSE.bin.txt
+++ b/distribution/server/src/assemble/LICENSE.bin.txt
@@ -540,9 +540,9 @@ BSD 2-Clause License
MIT License
* Java SemVer -- com.github.zafarkhaja-java-semver-0.9.0.jar -- licenses/LICENSE-SemVer.txt
* SLF4J -- licenses/LICENSE-SLF4J.txt
- - org.slf4j-jul-to-slf4j-1.7.25.jar
- - org.slf4j-slf4j-api-1.7.25.jar
- - org.slf4j-jcl-over-slf4j-1.7.25.jar
+ - org.slf4j-jul-to-slf4j-1.7.32.jar
+ - org.slf4j-slf4j-api-1.7.32.jar
+ - org.slf4j-jcl-over-slf4j-1.7.32.jar
* The Checker Framework
- org.checkerframework-checker-qual-3.5.0.jar
diff --git a/pom.xml b/pom.xml
index cfa1c34..2358fc7 100644
--- a/pom.xml
+++ b/pom.xml
@@ -117,7 +117,7 @@ flexible messaging model and an intuitive client API.</description>
<prometheus.version>0.5.0</prometheus.version>
<vertx.version>3.9.8</vertx.version>
<rocksdb.version>6.10.2</rocksdb.version>
- <slf4j.version>1.7.25</slf4j.version>
+ <slf4j.version>1.7.32</slf4j.version>
<commons.collections.version>3.2.2</commons.collections.version>
<log4j2.version>2.17.1</log4j2.version>
<bouncycastle.version>1.69</bouncycastle.version>
diff --git a/pulsar-sql/presto-distribution/LICENSE b/pulsar-sql/presto-distribution/LICENSE
index aaddf3a..d3ab8a0 100644
--- a/pulsar-sql/presto-distribution/LICENSE
+++ b/pulsar-sql/presto-distribution/LICENSE
@@ -485,12 +485,12 @@ MIT License
- pcollections-2.1.2.jar
* SLF4J
- slf4j-jdk14-1.7.29.jar
- - slf4j-api-1.7.25.jar
+ - slf4j-api-1.7.32.jar
- slf4j-jdk14-1.7.30.jar
* JCL 1.2 Implemented Over SLF4J
- - jcl-over-slf4j-1.7.25.jar
+ - jcl-over-slf4j-1.7.32.jar
* JUL to SLF4J Bridge
- - jul-to-slf4j-1.7.25.jar
+ - jul-to-slf4j-1.7.32.jar
* Checker Qual
- checker-qual-3.5.0.jar
[pulsar] 07/16: update log content (#13540)
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.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 9eef658d12bbe4f187765f324a63677a29047e35
Author: chenlin <15...@qq.com>
AuthorDate: Thu Jan 6 10:23:43 2022 +0800
update log content (#13540)
* update log content
* update log content
* check style
(cherry picked from commit 825d2fe236306242de639c38a0a60d7f3ec5ced0)
---
.../apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java
index 7668c11..7bcfcab 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java
@@ -607,7 +607,8 @@ public class ModularLoadManagerImpl implements ModularLoadManager {
return;
}
- log.info("[Overload shedder] Unloading bundle: {} from broker {}", bundle, broker);
+ log.info("[{}] Unloading bundle: {} from broker {}",
+ strategy.getClass().getSimpleName(), bundle, broker);
try {
pulsar.getAdminClient().namespaces().unloadNamespaceBundle(namespaceName, bundleRange);
loadData.getRecentlyUnloadedBundles().put(bundle, System.currentTimeMillis());
[pulsar] 16/16: [Java Client] fixed Producer semaphore permit release issue (#13682)
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.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit a08f0f506e5985df50a818e601ee4b3869e1ce02
Author: Qiang Zhao <74...@users.noreply.github.com>
AuthorDate: Mon Jan 10 19:27:26 2022 +0800
[Java Client] fixed Producer semaphore permit release issue (#13682)
(cherry picked from commit 916b61d3a497c44c8511347d2c4eac95c53f8506)
---
.../apache/pulsar/client/impl/ProducerSemaphoreTest.java | 14 ++++++++++++++
.../java/org/apache/pulsar/client/impl/ProducerImpl.java | 16 +++++++++++-----
2 files changed, 25 insertions(+), 5 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerSemaphoreTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerSemaphoreTest.java
index c719cbd..0c7f4b1 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerSemaphoreTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerSemaphoreTest.java
@@ -75,11 +75,13 @@ public class ProducerSemaphoreTest extends ProducerConsumerBase {
futures.add(producer.newMessage().value(("Semaphore-test-" + i).getBytes()).sendAsync());
}
Assert.assertEquals(producer.getSemaphore().get().availablePermits(), pendingQueueSize - messages);
+ Assert.assertFalse(producer.isErrorStat());
} finally {
producer.getClientCnx().channel().config().setAutoRead(true);
}
FutureUtil.waitForAll(futures).get();
Assert.assertEquals(producer.getSemaphore().get().availablePermits(), pendingQueueSize);
+ Assert.assertFalse(producer.isErrorStat());
futures.clear();
// Simulate replicator, non batching message but `numMessagesInBatch` of message metadata > 1
@@ -92,15 +94,18 @@ public class ProducerSemaphoreTest extends ProducerConsumerBase {
futures.add(producer.sendAsync(msg));
}
Assert.assertEquals(producer.getSemaphore().get().availablePermits(), pendingQueueSize - messages/2);
+ Assert.assertFalse(producer.isErrorStat());
} finally {
producer.getClientCnx().channel().config().setAutoRead(true);
}
FutureUtil.waitForAll(futures).get();
Assert.assertEquals(producer.getSemaphore().get().availablePermits(), pendingQueueSize);
+ Assert.assertFalse(producer.isErrorStat());
futures.clear();
// Here must ensure that the semaphore available permits is 0
Assert.assertEquals(producer.getSemaphore().get().availablePermits(), pendingQueueSize);
+ Assert.assertFalse(producer.isErrorStat());
// Acquire 5 and not wait the send ack call back
producer.getClientCnx().channel().config().setAutoRead(false);
@@ -111,12 +116,14 @@ public class ProducerSemaphoreTest extends ProducerConsumerBase {
// Here must ensure that the Semaphore a acquired 5
Assert.assertEquals(producer.getSemaphore().get().availablePermits(), pendingQueueSize - messages / 2);
+ Assert.assertFalse(producer.isErrorStat());
} finally {
producer.getClientCnx().channel().config().setAutoRead(true);
}
FutureUtil.waitForAll(futures).get();
Assert.assertEquals(producer.getSemaphore().get().availablePermits(), pendingQueueSize);
+ Assert.assertFalse(producer.isErrorStat());
}
/**
@@ -141,6 +148,7 @@ public class ProducerSemaphoreTest extends ProducerConsumerBase {
// Test that when we fill the queue with "replicator" messages, we are notified
// (replicator itself would block)
Assert.assertEquals(producer.getSemaphore().get().availablePermits(), pendingQueueSize);
+ Assert.assertFalse(producer.isErrorStat());
producer.getClientCnx().channel().config().setAutoRead(false);
try {
for (int i = 0; i < pendingQueueSize; i++) {
@@ -151,6 +159,7 @@ public class ProducerSemaphoreTest extends ProducerConsumerBase {
futures.add(producer.sendAsync(msg));
}
Assert.assertEquals(producer.getSemaphore().get().availablePermits(), 0);
+ Assert.assertFalse(producer.isErrorStat());
try {
MessageMetadata metadata = new MessageMetadata()
.setNumMessagesInBatch(10);
@@ -162,6 +171,7 @@ public class ProducerSemaphoreTest extends ProducerConsumerBase {
Assert.assertEquals(ee.getCause().getClass(),
PulsarClientException.ProducerQueueIsFullError.class);
Assert.assertEquals(producer.getSemaphore().get().availablePermits(), 0);
+ Assert.assertFalse(producer.isErrorStat());
}
} finally {
producer.getClientCnx().channel().config().setAutoRead(true);
@@ -171,12 +181,14 @@ public class ProducerSemaphoreTest extends ProducerConsumerBase {
// Test that when we fill the queue with normal messages, we get an error
Assert.assertEquals(producer.getSemaphore().get().availablePermits(), pendingQueueSize);
+ Assert.assertFalse(producer.isErrorStat());
producer.getClientCnx().channel().config().setAutoRead(false);
try {
for (int i = 0; i < pendingQueueSize; i++) {
futures.add(producer.newMessage().value(("Semaphore-test-" + i).getBytes()).sendAsync());
}
Assert.assertEquals(producer.getSemaphore().get().availablePermits(), 0);
+ Assert.assertFalse(producer.isErrorStat());
try {
producer.newMessage().value(("Semaphore-test-Q-full").getBytes()).sendAsync().get();
@@ -184,6 +196,7 @@ public class ProducerSemaphoreTest extends ProducerConsumerBase {
Assert.assertEquals(ee.getCause().getClass(),
PulsarClientException.ProducerQueueIsFullError.class);
Assert.assertEquals(producer.getSemaphore().get().availablePermits(), 0);
+ Assert.assertFalse(producer.isErrorStat());
}
} finally {
@@ -191,5 +204,6 @@ public class ProducerSemaphoreTest extends ProducerConsumerBase {
}
FutureUtil.waitForAll(futures).get();
Assert.assertEquals(producer.getSemaphore().get().availablePermits(), pendingQueueSize);
+ Assert.assertFalse(producer.isErrorStat());
}
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
index f5165d7..46649bd 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
@@ -266,11 +266,12 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
protected void semaphoreRelease(final int releaseCountRequest) {
if (semaphore.isPresent()) {
if (!errorState) {
- final int availablePermits = semaphore.get().availablePermits();
- if (availablePermits - releaseCountRequest < 0) {
- log.error("Semaphore permit release count request greater then availablePermits" +
- " : availablePermits={}, releaseCountRequest={}",
- availablePermits, releaseCountRequest);
+ final int availableReleasePermits =
+ conf.getMaxPendingMessages() - this.semaphore.get().availablePermits();
+ if (availableReleasePermits - releaseCountRequest < 0) {
+ log.error("Semaphore permit release count request greater then availableReleasePermits" +
+ " : availableReleasePermits={}, releaseCountRequest={}",
+ availableReleasePermits, releaseCountRequest);
errorState = true;
}
}
@@ -2043,5 +2044,10 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
return semaphore;
}
+ @VisibleForTesting
+ boolean isErrorStat() {
+ return errorState;
+ }
+
private static final Logger log = LoggerFactory.getLogger(ProducerImpl.class);
}
[pulsar] 13/16: Fix reader skipped remaining compacted data during the topic unloading. (#13629)
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.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 23700dbe7eaf871a2160a47546dc3b03f51f3e1e
Author: lipenghui <pe...@apache.org>
AuthorDate: Fri Jan 7 13:44:02 2022 +0800
Fix reader skipped remaining compacted data during the topic unloading. (#13629)
### Motivation
To fix the reader skipping remaining compacted data while the topic has been unloaded.
#11287 fixed the data skipped issue while the reader first time to read the messages
with the earliest position. But if the reader has consumed some messages from the
compacted ledger but not all, the start position will not be `earliest`, the broker
will rewind the cursor for the reader to the next valid position of the original topic.
So the remaining messages in the compacted ledger will be skipped.
Here are the logs from the broker:
```
10:44:36.035 [bookkeeper-ml-scheduler-OrderedScheduler-4-0] INFO org.apache.pulsar.broker.service.BrokerService - Created topic persistent://xxx/product-full-prod/5126 - dedup is disabled
10:44:36.035 [bookkeeper-ml-scheduler-OrderedScheduler-4-0] INFO org.apache.pulsar.broker.service.persistent.PersistentTopic - [persistent://xxx/product-full-prod/5126][xxx] Creating non-durable subscription at msg id 181759:14:-1:-1
10:44:36.035 [bookkeeper-ml-scheduler-OrderedScheduler-4-0] INFO org.apache.bookkeeper.mledger.impl.NonDurableCursorImpl - [xxx/product-full-prod/persistent/5126] Created non-durable cursor read-position=221199:0 mark-delete-position=181759:13
10:44:36.035 [bookkeeper-ml-scheduler-OrderedScheduler-4-0] INFO org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl - [xxx/product-full-prod/persistent/5126] Opened new cursor: NonDurableCursorImpl{ledger=xxx/product-full-prod/persistent/5126, ackPos=181759:13, readPos=221199:0}
10:44:36.035 [bookkeeper-ml-scheduler-OrderedScheduler-4-0] INFO org.apache.bookkeeper.mledger.impl.ManagedCursorImpl - [xxx/product-full-prod/persistent/5126-xxx] Rewind from 221199:0 to 221199:0
```
There some many compacted messages after `181759:13`, but the broker will not dispatch them to the reader.
The issue also can be reproduced by the unit test that was added in this PR.
### Modification
If the cursor with `readCompacted = true`, just rewind to the next message of the mark delete position,
so that the reader can continue to read the data from the compacted ledger.
### Verification
A new test added for testing the reader can get all the compacted messages and non-compacted messages from the topic during the topic unloading.
(cherry picked from commit 07f131fec7db36d6f424ce419d26888592b04207)
---
.../apache/bookkeeper/mledger/ManagedLedger.java | 3 +-
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 7 +--
.../mledger/impl/NonDurableCursorImpl.java | 23 ++++++---
.../AbstractDispatcherSingleActiveConsumer.java | 4 --
.../broker/service/persistent/PersistentTopic.java | 8 ++--
.../pulsar/compaction/CompactedTopicTest.java | 55 ++++++++++++++++++++++
.../offload/jcloud/impl/MockManagedLedger.java | 4 +-
7 files changed, 84 insertions(+), 20 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedger.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedger.java
index 74c67e9..0200e25 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedger.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedger.java
@@ -259,7 +259,8 @@ public interface ManagedLedger {
*/
ManagedCursor newNonDurableCursor(Position startCursorPosition) throws ManagedLedgerException;
ManagedCursor newNonDurableCursor(Position startPosition, String subscriptionName) throws ManagedLedgerException;
- ManagedCursor newNonDurableCursor(Position startPosition, String subscriptionName, InitialPosition initialPosition) throws ManagedLedgerException;
+ ManagedCursor newNonDurableCursor(Position startPosition, String subscriptionName, InitialPosition initialPosition,
+ boolean isReadCompacted) throws ManagedLedgerException;
/**
* Delete a ManagedCursor asynchronously.
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
index cec106c7..457600c 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
@@ -1030,11 +1030,12 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
@Override
public ManagedCursor newNonDurableCursor(Position startPosition, String subscriptionName) throws ManagedLedgerException {
- return newNonDurableCursor(startPosition, subscriptionName, InitialPosition.Latest);
+ return newNonDurableCursor(startPosition, subscriptionName, InitialPosition.Latest, false);
}
@Override
- public ManagedCursor newNonDurableCursor(Position startCursorPosition, String cursorName, InitialPosition initialPosition)
+ public ManagedCursor newNonDurableCursor(Position startCursorPosition, String cursorName, InitialPosition initialPosition,
+ boolean isReadCompacted)
throws ManagedLedgerException {
Objects.requireNonNull(cursorName, "cursor name can't be null");
checkManagedLedgerIsOpen();
@@ -1049,7 +1050,7 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
}
NonDurableCursorImpl cursor = new NonDurableCursorImpl(bookKeeper, config, this, cursorName,
- (PositionImpl) startCursorPosition, initialPosition);
+ (PositionImpl) startCursorPosition, initialPosition, isReadCompacted);
cursor.setActive();
log.info("[{}] Opened new cursor: {}", name, cursor);
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/NonDurableCursorImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/NonDurableCursorImpl.java
index 0f7ffe41..1d545bd 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/NonDurableCursorImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/NonDurableCursorImpl.java
@@ -33,11 +33,12 @@ import org.slf4j.LoggerFactory;
public class NonDurableCursorImpl extends ManagedCursorImpl {
- private volatile boolean readCompacted;
+ private final boolean readCompacted;
NonDurableCursorImpl(BookKeeper bookkeeper, ManagedLedgerConfig config, ManagedLedgerImpl ledger, String cursorName,
- PositionImpl startCursorPosition, CommandSubscribe.InitialPosition initialPosition) {
+ PositionImpl startCursorPosition, CommandSubscribe.InitialPosition initialPosition, boolean isReadCompacted) {
super(bookkeeper, config, ledger, cursorName);
+ this.readCompacted = isReadCompacted;
// Compare with "latest" position marker by using only the ledger id. Since the C++ client is using 48bits to
// store the entryId, it's not able to pass a Long.max() as entryId. In this case there's no point to require
@@ -67,7 +68,7 @@ public class NonDurableCursorImpl extends ManagedCursorImpl {
private void recoverCursor(PositionImpl mdPosition) {
Pair<PositionImpl, Long> lastEntryAndCounter = ledger.getLastPositionAndCounter();
- this.readPosition = ledger.getNextValidPosition(mdPosition);
+ this.readPosition = isReadCompacted() ? mdPosition.getNext() : ledger.getNextValidPosition(mdPosition);
markDeletePosition = mdPosition;
// Initialize the counter such that the difference between the messages written on the ML and the
@@ -118,15 +119,23 @@ public class NonDurableCursorImpl extends ManagedCursorImpl {
callback.deleteCursorComplete(ctx);
}
- public void setReadCompacted(boolean readCompacted) {
- this.readCompacted = readCompacted;
- }
-
public boolean isReadCompacted() {
return readCompacted;
}
@Override
+ public void rewind() {
+ // For reading the compacted data,
+ // we couldn't reset the read position to the next valid position of the original topic.
+ // Otherwise, the remaining data in the compacted ledger will be skipped.
+ if (!readCompacted) {
+ super.rewind();
+ } else {
+ readPosition = markDeletePosition.getNext();
+ }
+ }
+
+ @Override
public synchronized String toString() {
return MoreObjects.toStringHelper(this).add("ledger", ledger.getName()).add("ackPos", markDeletePosition)
.add("readPos", readPosition).toString();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java
index 4c7ea45..8cab06b 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java
@@ -27,7 +27,6 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import org.apache.bookkeeper.mledger.ManagedCursor;
-import org.apache.bookkeeper.mledger.impl.NonDurableCursorImpl;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.service.BrokerServiceException.ConsumerBusyException;
import org.apache.pulsar.broker.service.BrokerServiceException.ServerMetadataException;
@@ -181,9 +180,6 @@ public abstract class AbstractDispatcherSingleActiveConsumer extends AbstractBas
consumer.notifyActiveConsumerChange(currentActiveConsumer);
}
}
- if (cursor != null && !cursor.isDurable() && cursor instanceof NonDurableCursorImpl) {
- ((NonDurableCursorImpl) cursor).setReadCompacted(ACTIVE_CONSUMER_UPDATER.get(this).readCompacted());
- }
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index 1b2e11f..e2cc79c 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -745,7 +745,7 @@ public class PersistentTopic extends AbstractTopic
getDurableSubscription(subscriptionName, initialPosition, startMessageRollbackDurationSec,
replicatedSubscriptionState)
: getNonDurableSubscription(subscriptionName, startMessageId, initialPosition,
- startMessageRollbackDurationSec);
+ startMessageRollbackDurationSec, readCompacted);
int maxUnackedMessages = isDurable
? getMaxUnackedMessagesOnConsumer()
@@ -891,7 +891,8 @@ public class PersistentTopic extends AbstractTopic
}
private CompletableFuture<? extends Subscription> getNonDurableSubscription(String subscriptionName,
- MessageId startMessageId, InitialPosition initialPosition, long startMessageRollbackDurationSec) {
+ MessageId startMessageId, InitialPosition initialPosition, long startMessageRollbackDurationSec,
+ boolean isReadCompacted) {
log.info("[{}][{}] Creating non-durable subscription at msg id {}", topic, subscriptionName, startMessageId);
CompletableFuture<Subscription> subscriptionFuture = new CompletableFuture<>();
@@ -924,7 +925,8 @@ public class PersistentTopic extends AbstractTopic
Position startPosition = new PositionImpl(ledgerId, entryId);
ManagedCursor cursor = null;
try {
- cursor = ledger.newNonDurableCursor(startPosition, subscriptionName, initialPosition);
+ cursor = ledger.newNonDurableCursor(startPosition, subscriptionName, initialPosition,
+ isReadCompacted);
} catch (ManagedLedgerException e) {
return FutureUtil.failedFuture(e);
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicTest.java
index 4d00d28..accce4ef 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicTest.java
@@ -734,4 +734,59 @@ public class CompactedTopicTest extends MockedPulsarServiceBaseTest {
Assert.assertNull(reader.readNext(3, TimeUnit.SECONDS));
}
+ public void testReadCompleteMessagesDuringTopicUnloading() throws Exception {
+ String topic = "persistent://my-property/use/my-ns/testReadCompleteMessagesDuringTopicUnloading-" +
+ UUID.randomUUID();
+ final int numMessages = 1000;
+ @Cleanup
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic(topic)
+ .blockIfQueueFull(true)
+ .enableBatching(false)
+ .create();
+ CompletableFuture<MessageId> lastMessage = null;
+ for (int i = 0; i < numMessages; ++i) {
+ lastMessage = producer.newMessage().key(i + "").value(String.format("msg [%d]", i)).sendAsync();
+ }
+ producer.flush();
+ lastMessage.join();
+ admin.topics().triggerCompaction(topic);
+ Awaitility.await().untilAsserted(() -> {
+ PersistentTopicInternalStats stats = admin.topics().getInternalStats(topic);
+ Assert.assertNotEquals(stats.compactedLedger.ledgerId, -1);
+ Assert.assertEquals(stats.compactedLedger.entries, numMessages);
+ Assert.assertEquals(admin.topics().getStats(topic)
+ .getSubscriptions().get(COMPACTION_SUBSCRIPTION).getConsumers().size(), 0);
+ Assert.assertEquals(stats.lastConfirmedEntry, stats.cursors.get(COMPACTION_SUBSCRIPTION).markDeletePosition);
+ });
+ // Unload the topic to make sure the original ledger been deleted.
+ admin.topics().unload(topic);
+ // Produce more messages to the original topic
+ for (int i = 0; i < numMessages; ++i) {
+ lastMessage = producer.newMessage().key(i + numMessages + "").value(String.format("msg [%d]", i + numMessages)).sendAsync();
+ }
+ producer.flush();
+ lastMessage.join();
+ // For now the topic has 1000 messages in the compacted ledger and 1000 messages in the original topic.
+ @Cleanup
+ Reader<String> reader = pulsarClient.newReader(Schema.STRING)
+ .topic(topic)
+ .startMessageIdInclusive()
+ .startMessageId(MessageId.earliest)
+ .readCompacted(true)
+ .create();
+
+ // Unloading the topic during reading the data to make sure the reader will not miss any messages.
+ for (int i = 0; i < numMessages / 2; ++i) {
+ Assert.assertEquals(reader.readNext().getValue(), String.format("msg [%d]", i));
+ }
+ admin.topics().unload(topic);
+ for (int i = 0; i < numMessages / 2; ++i) {
+ Assert.assertEquals(reader.readNext().getValue(), String.format("msg [%d]", i + numMessages / 2));
+ }
+ admin.topics().unload(topic);
+ for (int i = 0; i < numMessages; ++i) {
+ Assert.assertEquals(reader.readNext().getValue(), String.format("msg [%d]", i + numMessages));
+ }
+ }
}
diff --git a/tiered-storage/jcloud/src/test/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/MockManagedLedger.java b/tiered-storage/jcloud/src/test/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/MockManagedLedger.java
index 767190c..229cd66 100644
--- a/tiered-storage/jcloud/src/test/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/MockManagedLedger.java
+++ b/tiered-storage/jcloud/src/test/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/MockManagedLedger.java
@@ -122,8 +122,8 @@ public class MockManagedLedger implements ManagedLedger {
@Override
public ManagedCursor newNonDurableCursor(Position startPosition, String subscriptionName,
- CommandSubscribe.InitialPosition initialPosition) throws
- ManagedLedgerException {
+ CommandSubscribe.InitialPosition initialPosition,
+ boolean isReadCompacted) throws ManagedLedgerException {
return null;
}
[pulsar] 12/16: [issue #13439][backlogQuota] Allow both limit and limitsize be null (#13557)
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.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 89c7b6e0d586812bd39bb2b21175dcad4c237316
Author: ZhangJian He <sh...@gmail.com>
AuthorDate: Fri Jan 7 15:14:45 2022 +0800
[issue #13439][backlogQuota] Allow both limit and limitsize be null (#13557)
Fixes #13439
### Motivation
Now we only constraint `limit` or `limitSize` be set on pulsar-admin, we don't limit it on rest endpoint. So there are situtations that both `limit` and `limitSize` be null
### Verifying this change
This change added tests and can be verified as follows:
- *Added test method in `BacklogQuotaCompatibilityTest`
(cherry picked from commit 33491bc2edf698723865d2be168e4b1082742d45)
---
.../common/policies/data/impl/BacklogQuotaImpl.java | 4 +---
.../metadata/BacklogQuotaCompatibilityTest.java | 20 ++++++++++++++++++++
2 files changed, 21 insertions(+), 3 deletions(-)
diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/impl/BacklogQuotaImpl.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/impl/BacklogQuotaImpl.java
index 3d97fa0..c8073d8 100644
--- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/impl/BacklogQuotaImpl.java
+++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/impl/BacklogQuotaImpl.java
@@ -35,7 +35,7 @@ public class BacklogQuotaImpl implements BacklogQuota {
* @since 2.9.1
*/
@Deprecated
- private Long limit;
+ private long limit;
/**
* backlog quota by size in byte.
@@ -57,7 +57,6 @@ public class BacklogQuotaImpl implements BacklogQuota {
@Deprecated
public long getLimit() {
if (limitSize == null) {
- // the limitSize and limit can't be both null
return limit;
}
return limitSize;
@@ -71,7 +70,6 @@ public class BacklogQuotaImpl implements BacklogQuota {
public long getLimitSize() {
if (limitSize == null) {
- // the limitSize and limit can't be both null
return limit;
}
return limitSize;
diff --git a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/BacklogQuotaCompatibilityTest.java b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/BacklogQuotaCompatibilityTest.java
index 659825c..f621a3a 100644
--- a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/BacklogQuotaCompatibilityTest.java
+++ b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/BacklogQuotaCompatibilityTest.java
@@ -105,4 +105,24 @@ public class BacklogQuotaCompatibilityTest {
BacklogQuota.RetentionPolicy.consumer_backlog_eviction);
}
+ @Test
+ public void testBackwardCompatibilityNullLimitAndLimitSize() throws IOException {
+ String oldPolicyStr = "{\"auth_policies\":{\"namespace_auth\":{},\"destination_auth\":{},"
+ + "\"subscription_auth_roles\":{}},\"replication_clusters\":[],\"backlog_quota_map\":"
+ + "{\"destination_storage\":{\"policy\":\"consumer_backlog_eviction\"}},"
+ + "\"clusterDispatchRate\":{},\"topicDispatchRate\":{},\"subscriptionDispatchRate\":{},"
+ + "\"replicatorDispatchRate\":{},\"clusterSubscribeRate\":{},\"publishMaxMessageRate\":{},"
+ + "\"latency_stats_sample_rate\":{},\"subscription_expiration_time_minutes\":0,\"deleted\":false,"
+ + "\"encryption_required\":false,\"subscription_auth_mode\":\"None\","
+ + "\"max_consumers_per_subscription\":0,\"offload_threshold\":-1,"
+ + "\"schema_auto_update_compatibility_strategy\":\"Full\",\"schema_compatibility_strategy\":"
+ + "\"UNDEFINED\",\"is_allow_auto_update_schema\":true,\"schema_validation_enforced\":false,"
+ + "\"subscription_types_enabled\":[]}\n";
+ Policies policies = simpleType.deserialize(null, oldPolicyStr.getBytes(), null);
+ assertEquals(policies.backlog_quota_map.get(BacklogQuota.BacklogQuotaType.destination_storage).getLimitSize(),
+ 0);
+ assertEquals(policies.backlog_quota_map.get(BacklogQuota.BacklogQuotaType.destination_storage).getLimitTime(),
+ 0);
+ }
+
}
[pulsar] 05/16: [Presto] Fix issue of mixed SLF4J library versions in presto libs (#13603)
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.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit f1f2f3dabe536a3fdeb2f7f51e93dd43c46f827a
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Tue Jan 4 20:26:54 2022 +0200
[Presto] Fix issue of mixed SLF4J library versions in presto libs (#13603)
(cherry picked from commit 29840b4ecabd0d997c25d52f6c02086b215f2cae)
---
pulsar-sql/pom.xml | 12 ++++++++++++
pulsar-sql/presto-distribution/LICENSE | 6 ++----
2 files changed, 14 insertions(+), 4 deletions(-)
diff --git a/pulsar-sql/pom.xml b/pulsar-sql/pom.xml
index 279b6a1..0261117 100644
--- a/pulsar-sql/pom.xml
+++ b/pulsar-sql/pom.xml
@@ -133,6 +133,18 @@
<artifactId>okio</artifactId>
<version>${okio.version}</version>
</dependency>
+
+ <!-- force specific version of slf4j -->
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>log4j-over-slf4j</artifactId>
+ <version>${slf4j.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-jdk14</artifactId>
+ <version>${slf4j.version}</version>
+ </dependency>
</dependencies>
</dependencyManagement>
diff --git a/pulsar-sql/presto-distribution/LICENSE b/pulsar-sql/presto-distribution/LICENSE
index d3ab8a0..a988218 100644
--- a/pulsar-sql/presto-distribution/LICENSE
+++ b/pulsar-sql/presto-distribution/LICENSE
@@ -329,8 +329,7 @@ The Apache Software License, Version 2.0
- leveldb-0.10.jar
- leveldb-api-0.10.jar
* Log4j implemented over SLF4J
- - log4j-over-slf4j-1.7.29.jar
- - log4j-over-slf4j-1.7.30.jar
+ - log4j-over-slf4j-1.7.32.jar
* Lucene Common Analyzers
- lucene-analyzers-common-8.4.1.jar
- lucene-core-8.4.1.jar
@@ -484,9 +483,8 @@ MIT License
* PCollections
- pcollections-2.1.2.jar
* SLF4J
- - slf4j-jdk14-1.7.29.jar
- slf4j-api-1.7.32.jar
- - slf4j-jdk14-1.7.30.jar
+ - slf4j-jdk14-1.7.32.jar
* JCL 1.2 Implemented Over SLF4J
- jcl-over-slf4j-1.7.32.jar
* JUL to SLF4J Bridge
[pulsar] 08/16: Fix NPE when unloading namespace bundle (#13571)
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.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 6fd742139ddfed161ea0c76582b7dfdd4be7004e
Author: feynmanlin <fe...@tencent.com>
AuthorDate: Wed Jan 5 14:46:32 2022 +0800
Fix NPE when unloading namespace bundle (#13571)
* Fix NPE when unloading namespace bundle
(cherry picked from commit ec0a44058d249a7510bb3d05685b2ee5e0874eb6)
---
.../apache/pulsar/broker/service/BrokerService.java | 18 ++++++++++--------
1 file changed, 10 insertions(+), 8 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index cbe02c8..d37dfb6 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -830,15 +830,17 @@ public class BrokerService implements Closeable {
// unload all namespace-bundles gracefully
long closeTopicsStartTime = System.nanoTime();
Set<NamespaceBundle> serviceUnits = pulsar.getNamespaceService().getOwnedServiceUnits();
- serviceUnits.forEach(su -> {
- if (su instanceof NamespaceBundle) {
- try {
- pulsar.getNamespaceService().unloadNamespaceBundle(su, 1, TimeUnit.MINUTES).get();
- } catch (Exception e) {
- log.warn("Failed to unload namespace bundle {}", su, e);
+ if (serviceUnits != null) {
+ serviceUnits.forEach(su -> {
+ if (su instanceof NamespaceBundle) {
+ try {
+ pulsar.getNamespaceService().unloadNamespaceBundle(su, 1, TimeUnit.MINUTES).get();
+ } catch (Exception e) {
+ log.warn("Failed to unload namespace bundle {}", su, e);
+ }
}
- }
- });
+ });
+ }
double closeTopicsTimeSeconds = TimeUnit.NANOSECONDS.toMillis((System.nanoTime() - closeTopicsStartTime))
/ 1000.0;
[pulsar] 14/16: Update notice year. (#13653)
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.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit d0904136da4654834248c30730fcaef6da59b270
Author: Jiwei Guo <te...@apache.org>
AuthorDate: Sat Jan 8 03:30:07 2022 +0800
Update notice year. (#13653)
### Motivation
Update NOTICE from 2021-2022
(cherry picked from commit bacac04f564536be9c30481d2cab88512906967e)
---
NOTICE | 2 +-
distribution/server/src/assemble/NOTICE.bin.txt | 2 +-
2 files changed, 2 insertions(+), 2 deletions(-)
diff --git a/NOTICE b/NOTICE
index 2052417..bbbe4fa 100644
--- a/NOTICE
+++ b/NOTICE
@@ -1,5 +1,5 @@
Apache Pulsar
-Copyright 2017-2021 The Apache Software Foundation
+Copyright 2017-2022 The Apache Software Foundation
This product includes software developed at
The Apache Software Foundation (http://www.apache.org/).
diff --git a/distribution/server/src/assemble/NOTICE.bin.txt b/distribution/server/src/assemble/NOTICE.bin.txt
index 661921c..bc5e2e6 100644
--- a/distribution/server/src/assemble/NOTICE.bin.txt
+++ b/distribution/server/src/assemble/NOTICE.bin.txt
@@ -1,6 +1,6 @@
Apache Pulsar
-Copyright 2017-2021 The Apache Software Foundation
+Copyright 2017-2022 The Apache Software Foundation
This product includes software developed at
The Apache Software Foundation (http://www.apache.org/).