coderzc · coderzc · Oct 13, 2022 · Oct 14, 2022 · Oct 14, 2022 · Oct 14, 2022
diff --git a/.github/PULL_REQUEST_TEMPLATE.md b/.github/PULL_REQUEST_TEMPLATE.md
@@ -56,6 +56,7 @@ This change added tests and can be verified as follows:
 - [ ] The public API
 - [ ] The schema
 - [ ] The default values of configurations
+- [ ] The threading model
 - [ ] The binary protocol
 - [ ] The REST endpoints
 - [ ] The admin CLI options

diff --git a/.github/workflows/ci-go-functions.yaml b/.github/workflows/ci-go-functions.yaml
@@ -54,7 +54,7 @@ jobs:
       - name: Check changed files
         id: check_changes
         run: |
-          echo "::set-output name=docs_only::${{ fromJSON(steps.changes.outputs.all_count) == fromJSON(steps.changes.outputs.docs_count) && fromJSON(steps.changes.outputs.docs_count) > 0 }}"
+          echo "docs_only=${{ fromJSON(steps.changes.outputs.all_count) == fromJSON(steps.changes.outputs.docs_count) && fromJSON(steps.changes.outputs.docs_count) > 0 }}" >> $GITHUB_OUTPUT
 
       - name: Check if the PR has been approved for testing
         if: ${{ steps.check_changes.outputs.docs_only != 'true' && github.repository == 'apache/pulsar' && github.event_name == 'pull_request' }}

diff --git a/.github/workflows/pulsar-ci-flaky.yaml b/.github/workflows/pulsar-ci-flaky.yaml
@@ -56,7 +56,7 @@ jobs:
       - name: Check changed files
         id: check_changes
         run: |
-          echo "::set-output name=docs_only::${{ fromJSON(steps.changes.outputs.all_count) == fromJSON(steps.changes.outputs.docs_count) && fromJSON(steps.changes.outputs.docs_count) > 0 }}"
+          echo "docs_only=${{ fromJSON(steps.changes.outputs.all_count) == fromJSON(steps.changes.outputs.docs_count) && fromJSON(steps.changes.outputs.docs_count) > 0 }}" >> $GITHUB_OUTPUT
 
       - name: Check if the PR has been approved for testing
         if: ${{ steps.check_changes.outputs.docs_only != 'true' && github.repository == 'apache/pulsar' && github.event_name == 'pull_request' }}

diff --git a/.github/workflows/pulsar-ci.yaml b/.github/workflows/pulsar-ci.yaml
@@ -56,7 +56,7 @@ jobs:
       - name: Check changed files
         id: check_changes
         run: |
-          echo "::set-output name=docs_only::${{ fromJSON(steps.changes.outputs.all_count) == fromJSON(steps.changes.outputs.docs_count) && fromJSON(steps.changes.outputs.docs_count) > 0 }}"
+          echo "docs_only=${{ fromJSON(steps.changes.outputs.all_count) == fromJSON(steps.changes.outputs.docs_count) && fromJSON(steps.changes.outputs.docs_count) > 0 }}" >> $GITHUB_OUTPUT
 
 
       - name: Check if the PR has been approved for testing
@@ -218,6 +218,7 @@ jobs:
         uses: ./.github/actions/copy-test-reports
 
       - name: Upload to Codecov
+        if: ${{ github.repository == 'apache/pulsar' }}
         uses: codecov/codecov-action@v3
         continue-on-error: true
         with:

diff --git a/README.md b/README.md
@@ -73,6 +73,8 @@ components in the Pulsar ecosystem, including connectors, adapters, and other la
 
 ### Clients
 
+- [C++ Client](https://github.com/apache/pulsar-client-cpp)
+- [Python Client](https://github.com/apache/pulsar-client-python)
 - [.NET/C# Client](https://github.com/apache/pulsar-dotpulsar)
 - [Go Client](https://github.com/apache/pulsar-client-go)
 - [NodeJS Client](https://github.com/apache/pulsar-client-node)

diff --git a/distribution/server/src/assemble/LICENSE.bin.txt b/distribution/server/src/assemble/LICENSE.bin.txt
@@ -246,14 +246,14 @@ The Apache Software License, Version 2.0
  * JCommander -- com.beust-jcommander-1.82.jar
  * High Performance Primitive Collections for Java -- com.carrotsearch-hppc-0.9.1.jar
  * Jackson
-     - com.fasterxml.jackson.core-jackson-annotations-2.13.3.jar
-     - com.fasterxml.jackson.core-jackson-core-2.13.3.jar
-     - com.fasterxml.jackson.core-jackson-databind-2.13.3.jar
-     - com.fasterxml.jackson.dataformat-jackson-dataformat-yaml-2.13.3.jar
-     - com.fasterxml.jackson.jaxrs-jackson-jaxrs-base-2.13.3.jar
-     - com.fasterxml.jackson.jaxrs-jackson-jaxrs-json-provider-2.13.3.jar
-     - com.fasterxml.jackson.module-jackson-module-jaxb-annotations-2.13.3.jar
-     - com.fasterxml.jackson.module-jackson-module-jsonSchema-2.13.3.jar
+     - com.fasterxml.jackson.core-jackson-annotations-2.13.4.jar
+     - com.fasterxml.jackson.core-jackson-core-2.13.4.jar
+     - com.fasterxml.jackson.core-jackson-databind-2.13.4.jar
+     - com.fasterxml.jackson.dataformat-jackson-dataformat-yaml-2.13.4.jar
+     - com.fasterxml.jackson.jaxrs-jackson-jaxrs-base-2.13.4.jar
+     - com.fasterxml.jackson.jaxrs-jackson-jaxrs-json-provider-2.13.4.jar
+     - com.fasterxml.jackson.module-jackson-module-jaxb-annotations-2.13.4.jar
+     - com.fasterxml.jackson.module-jackson-module-jsonSchema-2.13.4.jar
  * Caffeine -- com.github.ben-manes.caffeine-caffeine-2.9.1.jar
  * Conscrypt -- org.conscrypt-conscrypt-openjdk-uber-2.5.2.jar
  * Proto Google Common Protos -- com.google.api.grpc-proto-google-common-protos-2.0.1.jar
@@ -284,7 +284,7 @@ The Apache Software License, Version 2.0
     - org.apache.commons-commons-collections4-4.4.jar
     - org.apache.commons-commons-compress-1.21.jar
     - org.apache.commons-commons-lang3-3.11.jar
-    - org.apache.commons-commons-text-1.9.jar
+    - org.apache.commons-commons-text-1.10.0.jar
  * Netty
     - io.netty-netty-buffer-4.1.77.Final.jar
     - io.netty-netty-codec-4.1.77.Final.jar

diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerBkTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerBkTest.java
@@ -18,6 +18,7 @@
  */
 package org.apache.bookkeeper.mledger.impl;
 
+import static org.apache.pulsar.common.util.PortManager.releaseLockedPort;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertNotNull;
 import static org.testng.Assert.assertTrue;
@@ -117,7 +118,7 @@ public void testBookieFailure() throws Exception {
         metadataStore.unsetAlwaysFail();
 
         bkc = new BookKeeperTestClient(baseClientConf);
-        startNewBookie();
+        int port = startNewBookie();
 
         // Reconnect a new bk client
         factory.shutdown();
@@ -147,6 +148,7 @@ public void testBookieFailure() throws Exception {
         assertEquals("entry-2", new String(entries.get(0).getData()));
         entries.forEach(Entry::release);
         factory.shutdown();
+        releaseLockedPort(port);
     }
 
     @Test

diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/test/BookKeeperClusterTestCase.java b/managed-ledger/src/test/java/org/apache/bookkeeper/test/BookKeeperClusterTestCase.java
@@ -24,6 +24,7 @@
 package org.apache.bookkeeper.test;
 
 import static org.apache.bookkeeper.util.BookKeeperConstants.AVAILABLE_NODE;
+import static org.apache.pulsar.common.util.PortManager.nextLockedFreePort;
 import static org.testng.Assert.assertFalse;
 
 import com.google.common.base.Stopwatch;
@@ -62,7 +63,7 @@
 import org.apache.bookkeeper.proto.BookieServer;
 import org.apache.bookkeeper.replication.Auditor;
 import org.apache.bookkeeper.replication.ReplicationWorker;
-import org.apache.bookkeeper.util.PortManager;
+import org.apache.pulsar.common.util.PortManager;
 import org.apache.pulsar.metadata.api.MetadataStoreConfig;
 import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
 import org.apache.pulsar.metadata.impl.FaultInjectionMetadataStore;
@@ -113,6 +114,7 @@ public void handleTestMethodName(Method method) {
 
     private boolean isAutoRecoveryEnabled;
     protected ExecutorService executor;
+    private final List<Integer> bookiePorts = new ArrayList<>();
 
     SynchronousQueue<Throwable> asyncExceptions = new SynchronousQueue<>();
     protected void captureThrowable(Runnable c) {
@@ -264,7 +266,7 @@ protected void startBKCluster(String metadataServiceUri) throws Exception {
 
         // Create Bookie Servers (B1, B2, B3)
         for (int i = 0; i < numBookies; i++) {
-            startNewBookie();
+            bookiePorts.add(startNewBookie());
         }
     }
 
@@ -283,14 +285,15 @@ protected void stopBKCluster() throws Exception {
             t.shutdown();
         }
         servers.clear();
+        bookiePorts.removeIf(PortManager::releaseLockedPort);
     }
 
     protected ServerConfiguration newServerConfiguration() throws Exception {
         File f = tmpDirs.createNew("bookie", "test");
 
         int port;
         if (baseConf.isEnableLocalTransport() || !baseConf.getAllowEphemeralPorts()) {
-            port = PortManager.nextFreePort();
+            port = nextLockedFreePort();
         } else {
             port = 0;
         }

diff --git a/pom.xml b/pom.xml
@@ -121,7 +121,7 @@ flexible messaging model and an intuitive client API.</description>
     <bookkeeper.version>4.15.1</bookkeeper.version>
     <zookeeper.version>3.8.0</zookeeper.version>
     <commons-cli.version>1.5.0</commons-cli.version>
-    <commons-text.version>1.9</commons-text.version>
+    <commons-text.version>1.10.0</commons-text.version>
     <snappy.version>1.1.8.4</snappy.version> <!-- ZooKeeper server -->
     <dropwizardmetrics.version>4.1.12.1</dropwizardmetrics.version> <!-- ZooKeeper server -->
     <curator.version>5.1.0</curator.version>
@@ -139,7 +139,7 @@ flexible messaging model and an intuitive client API.</description>
     <log4j2.version>2.18.0</log4j2.version>
     <bouncycastle.version>1.69</bouncycastle.version>
     <bouncycastlefips.version>1.0.2</bouncycastlefips.version>
-    <jackson.version>2.13.3</jackson.version>
+    <jackson.version>2.13.4</jackson.version>
     <reflections.version>0.9.11</reflections.version>
     <swagger.version>1.6.2</swagger.version>
     <puppycrawl.checkstyle.version>8.37</puppycrawl.checkstyle.version>
@@ -247,6 +247,7 @@ flexible messaging model and an intuitive client API.</description>
     <objenesis.version>3.1</objenesis.version>
     <awaitility.version>4.2.0</awaitility.version>
     <reload4j.version>1.2.22</reload4j.version>
+    <jettison.version>1.5.1</jettison.version>
 
     <!-- Plugin dependencies -->
     <protobuf-maven-plugin.version>0.6.1</protobuf-maven-plugin.version>
@@ -798,6 +799,13 @@ flexible messaging model and an intuitive client API.</description>
         <scope>import</scope>
       </dependency>
 
+      <dependency>
+        <groupId>org.codehaus.jettison</groupId>
+        <artifactId>jettison</artifactId>
+        <version>${jettison.version}</version>
+      </dependency>
+
+
       <dependency>
         <groupId>org.hdrhistogram</groupId>
         <artifactId>HdrHistogram</artifactId>

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
@@ -4524,7 +4524,7 @@ private CompletableFuture<Void> createSubscriptions(TopicName topicName, int num
                 });
 
                 FutureUtil.waitForAll(subscriptionFutures).thenRun(() -> {
-                    log.info("[{}] Successfully created new partitions {}", clientAppId(), topicName);
+                    log.info("[{}] Successfully created subscriptions on new partitions {}", clientAppId(), topicName);
                     result.complete(null);
                 }).exceptionally(ex -> {
                     log.warn("[{}] Failed to create subscriptions on new partitions for {}",

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java
@@ -2054,7 +2054,8 @@ public void getOffloadThreshold(
         validateNamespacePolicyOperationAsync(namespaceName, PolicyName.OFFLOAD, PolicyOperation.READ)
                 .thenCompose(__ -> getNamespacePoliciesAsync(namespaceName))
                 .thenAccept(policies -> {
-                    if (policies.offload_policies == null) {
+                    if (policies.offload_policies == null
+                            || policies.offload_policies.getManagedLedgerOffloadThresholdInBytes() == null) {
                         asyncResponse.resume(policies.offload_threshold);
                     } else {
                         asyncResponse.resume(policies.offload_policies.getManagedLedgerOffloadThresholdInBytes());

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/BrokerOperabilityMetrics.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/BrokerOperabilityMetrics.java
@@ -32,8 +32,6 @@ public class BrokerOperabilityMetrics {
     private final List<Metrics> metricsList;
     private final String localCluster;
     private final DimensionStats topicLoadStats;
-    private final DimensionStats zkWriteLatencyStats;
-    private final DimensionStats zkReadLatencyStats;
     private final String brokerName;
     private final LongAdder connectionTotalCreatedCount;
     private final LongAdder connectionCreateSuccessCount;
@@ -45,8 +43,6 @@ public BrokerOperabilityMetrics(String localCluster, String brokerName) {
         this.metricsList = new ArrayList<>();
         this.localCluster = localCluster;
         this.topicLoadStats = new DimensionStats("topic_load_times", 60);
-        this.zkWriteLatencyStats = new DimensionStats("zk_write_latency", 60);
-        this.zkReadLatencyStats = new DimensionStats("zk_read_latency", 60);
         this.brokerName = brokerName;
         this.connectionTotalCreatedCount = new LongAdder();
         this.connectionCreateSuccessCount = new LongAdder();
@@ -62,8 +58,6 @@ public List<Metrics> getMetrics() {
 
     private void generate() {
         metricsList.add(getTopicLoadMetrics());
-        metricsList.add(getZkWriteLatencyMetrics());
-        metricsList.add(getZkReadLatencyMetrics());
         metricsList.add(getConnectionMetrics());
     }
 
@@ -93,14 +87,6 @@ Metrics getTopicLoadMetrics() {
         return getDimensionMetrics("topic_load_times", "topic_load", topicLoadStats);
     }
 
-    Metrics getZkWriteLatencyMetrics() {
-        return getDimensionMetrics("zk_write_latency", "zk_write", zkWriteLatencyStats);
-    }
-
-    Metrics getZkReadLatencyMetrics() {
-        return getDimensionMetrics("zk_read_latency", "zk_read", zkReadLatencyStats);
-    }
-
     Metrics getDimensionMetrics(String metricsName, String dimensionName, DimensionStats stats) {
         Metrics dMetrics = Metrics.create(getDimensionMap(metricsName));
 
@@ -120,22 +106,12 @@ Metrics getDimensionMetrics(String metricsName, String dimensionName, DimensionS
     public void reset() {
         metricsList.clear();
         topicLoadStats.reset();
-        zkWriteLatencyStats.reset();
-        zkReadLatencyStats.reset();
     }
 
     public void recordTopicLoadTimeValue(long topicLoadLatencyMs) {
         topicLoadStats.recordDimensionTimeValue(topicLoadLatencyMs, TimeUnit.MILLISECONDS);
     }
 
-    public void recordZkWriteLatencyTimeValue(long topicLoadLatencyMs) {
-        zkWriteLatencyStats.recordDimensionTimeValue(topicLoadLatencyMs, TimeUnit.MILLISECONDS);
-    }
-
-    public void recordZkReadLatencyTimeValue(long topicLoadLatencyMs) {
-        zkReadLatencyStats.recordDimensionTimeValue(topicLoadLatencyMs, TimeUnit.MILLISECONDS);
-    }
-
     public void recordConnectionCreate() {
         this.connectionTotalCreatedCount.increment();
         this.connectionActive.increment();

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/MultiBrokerBaseTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/MultiBrokerBaseTest.java
@@ -21,12 +21,14 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+
 import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
 import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.client.admin.PulsarAdminBuilder;
 import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.common.util.PortManager;
 import org.apache.pulsar.metadata.api.MetadataStoreException;
 import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
 import org.apache.pulsar.metadata.impl.ZKMetadataStore;
@@ -124,6 +126,9 @@ protected void additionalBrokersCleanup() {
                 try {
                     pulsarService.getConfiguration().setBrokerShutdownTimeoutMs(0L);
                     pulsarService.close();
+                    pulsarService.getConfiguration().getBrokerServicePort().ifPresent(PortManager::releaseLockedPort);
+                    pulsarService.getConfiguration().getWebServicePort().ifPresent(PortManager::releaseLockedPort);
+                    pulsarService.getConfiguration().getWebServicePortTls().ifPresent(PortManager::releaseLockedPort);
                 } catch (PulsarServerException e) {
                     // ignore
                 }

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java
@@ -164,7 +164,8 @@ public void resetClusters() throws Exception {
         pulsar.getConfiguration().setForceDeleteNamespaceAllowed(true);
         for (String tenant : admin.tenants().getTenants()) {
             for (String namespace : admin.namespaces().getNamespaces(tenant)) {
-                deleteNamespaceGraceFully(namespace, true);
+                deleteNamespaceGraceFullyByMultiPulsars(namespace, true, admin, pulsar,
+                        mockPulsarSetup.getPulsar());
             }
             admin.tenants().deleteTenant(tenant, true);
         }

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java
@@ -1401,6 +1401,9 @@ public void testSetOffloadThreshold() throws Exception {
         admin.namespaces().createNamespace(namespace, Set.of(testLocalCluster));
         admin.topics().createNonPartitionedTopic(topicName.toString());
 
+        admin.namespaces().setOffloadDeleteLag(namespace, 10000, TimeUnit.SECONDS);
+        assertEquals(-1, admin.namespaces().getOffloadThreshold(namespace));
+
         // assert we get the default which indicates it will fall back to default
         assertEquals(-1, admin.namespaces().getOffloadThreshold(namespace));
         // the ledger config should have the expected value