[improve][broker] Sync commits from apache/pulsar branch-3.1 into 3.1_ds (2)

build/run_unit_group.sh

@@ -135,13 +135,21 @@ function print_testng_failures() {
 function test_group_broker_flaky() {
   echo "::endgroup::"
   echo "::group::Running quarantined tests"
-  mvn_test --no-fail-fast -pl pulsar-broker -Dgroups='quarantine' -DexcludedGroups='' -DfailIfNoTests=false \
+  mvn_test --no-fail-fast -pl pulsar-broker -Dgroups='quarantine' -DexcludedGroups='flaky' -DfailIfNoTests=false \
     -DtestForkCount=2 ||
     print_testng_failures pulsar-broker/target/surefire-reports/testng-failed.xml "Quarantined test failure in" "Quarantined test failures"
   echo "::endgroup::"
   echo "::group::Running flaky tests"
-  mvn_test --no-fail-fast -pl pulsar-broker -Dgroups='flaky' -DtestForkCount=2
+  mvn_test --no-fail-fast -pl pulsar-broker -Dgroups='flaky' -DexcludedGroups='quarantine' -DtestForkCount=2
   echo "::endgroup::"
+  local modules_with_flaky_tests=$(git grep -l '@Test.*"flaky"' | grep '/src/test/java/' | \
+    awk -F '/src/test/java/' '{ print $1 }' | grep -v -E 'pulsar-broker' | sort | uniq | \
+    perl -0777 -p -e 's/\n(\S)/,$1/g')
+  if [ -n "${modules_with_flaky_tests}" ]; then
+    echo "::group::Running flaky tests in modules '${modules_with_flaky_tests}'"
+    mvn_test --no-fail-fast -pl "${modules_with_flaky_tests}" -Dgroups='flaky' -DexcludedGroups='quarantine' -DfailIfNoTests=false
+    echo "::endgroup::"
+  fi
 }
 
 function test_group_proxy() {
@@ -175,7 +183,7 @@ function test_group_other() {
     perl -0777 -p -e 's/\n(\S)/,$1/g')
   if [ -n "${modules_with_quarantined_tests}" ]; then
     echo "::group::Running quarantined tests outside of pulsar-broker & pulsar-proxy (if any)"
-    mvn_test --no-fail-fast -pl "${modules_with_quarantined_tests}" test -Dgroups='quarantine' -DexcludedGroups='' \
+    mvn_test --no-fail-fast -pl "${modules_with_quarantined_tests}" test -Dgroups='quarantine' -DexcludedGroups='flaky' \
       -DfailIfNoTests=false || \
         echo "::warning::There were test failures in the 'quarantine' test group."
     echo "::endgroup::"

buildtools/pom.xml

@@ -47,7 +47,7 @@
     <license-maven-plugin.version>4.1</license-maven-plugin.version>
     <puppycrawl.checkstyle.version>10.14.2</puppycrawl.checkstyle.version>
     <maven-checkstyle-plugin.version>3.1.2</maven-checkstyle-plugin.version>
-    <netty.version>4.1.100.Final</netty.version>
+    <netty.version>4.1.108.Final</netty.version>
     <guice.version>4.2.3</guice.version>
     <guava.version>32.1.1-jre</guava.version>
     <ant.version>1.10.12</ant.version>

buildtools/src/main/java/org/apache/pulsar/tests/AnnotationListener.java

@@ -32,6 +32,10 @@ public class AnnotationListener implements IAnnotationTransformer {
     private static final long DEFAULT_TEST_TIMEOUT_MILLIS = TimeUnit.MINUTES.toMillis(5);
     private static final String OTHER_GROUP = "other";
 
+    private static final String FLAKY_GROUP = "flaky";
+
+    private static final String QUARANTINE_GROUP = "quarantine";
+
     public AnnotationListener() {
         System.out.println("Created annotation listener");
     }
@@ -51,9 +55,27 @@ public void transform(ITestAnnotation annotation,
             annotation.setTimeOut(DEFAULT_TEST_TIMEOUT_MILLIS);
         }
 
+        replaceGroupsIfFlakyOrQuarantineGroupIsIncluded(annotation);
         addToOtherGroupIfNoGroupsSpecified(annotation);
     }
 
+    // A test method will inherit the test groups from the class level and this solution ensures that a test method
+    // added to the flaky or quarantine group will not be executed as part of other groups.
+    private void replaceGroupsIfFlakyOrQuarantineGroupIsIncluded(ITestAnnotation annotation) {
+        if (annotation.getGroups() != null && annotation.getGroups().length > 1) {
+            for (String group : annotation.getGroups()) {
+                if (group.equals(QUARANTINE_GROUP)) {
+                    annotation.setGroups(new String[]{QUARANTINE_GROUP});
+                    return;
+                }
+                if (group.equals(FLAKY_GROUP)) {
+                    annotation.setGroups(new String[]{FLAKY_GROUP});
+                    return;
+                }
+            }
+        }
+    }
+
     private void addToOtherGroupIfNoGroupsSpecified(ITestOrConfiguration annotation) {
         // Add test to "other" group if there's no specified group
         if (annotation.getGroups() == null || annotation.getGroups().length == 0) {

distribution/server/src/assemble/LICENSE.bin.txt

@@ -289,33 +289,33 @@ The Apache Software License, Version 2.0
     - org.apache.commons-commons-lang3-3.11.jar
     - org.apache.commons-commons-text-1.10.0.jar
  * Netty
-    - io.netty-netty-buffer-4.1.100.Final.jar
-    - io.netty-netty-codec-4.1.100.Final.jar
-    - io.netty-netty-codec-dns-4.1.100.Final.jar
-    - io.netty-netty-codec-http-4.1.100.Final.jar
-    - io.netty-netty-codec-http2-4.1.100.Final.jar
-    - io.netty-netty-codec-socks-4.1.100.Final.jar
-    - io.netty-netty-codec-haproxy-4.1.100.Final.jar
-    - io.netty-netty-common-4.1.100.Final.jar
-    - io.netty-netty-handler-4.1.100.Final.jar
-    - io.netty-netty-handler-proxy-4.1.100.Final.jar
-    - io.netty-netty-resolver-4.1.100.Final.jar
-    - io.netty-netty-resolver-dns-4.1.100.Final.jar
-    - io.netty-netty-resolver-dns-classes-macos-4.1.100.Final.jar
-    - io.netty-netty-resolver-dns-native-macos-4.1.100.Final-osx-aarch_64.jar
-    - io.netty-netty-resolver-dns-native-macos-4.1.100.Final-osx-x86_64.jar
-    - io.netty-netty-transport-4.1.100.Final.jar
-    - io.netty-netty-transport-classes-epoll-4.1.100.Final.jar
-    - io.netty-netty-transport-native-epoll-4.1.100.Final-linux-x86_64.jar
-    - io.netty-netty-transport-native-unix-common-4.1.100.Final.jar
-    - io.netty-netty-transport-native-unix-common-4.1.100.Final-linux-x86_64.jar
-    - io.netty-netty-tcnative-boringssl-static-2.0.61.Final.jar
-    - io.netty-netty-tcnative-boringssl-static-2.0.61.Final-linux-aarch_64.jar
-    - io.netty-netty-tcnative-boringssl-static-2.0.61.Final-linux-x86_64.jar
-    - io.netty-netty-tcnative-boringssl-static-2.0.61.Final-osx-aarch_64.jar
-    - io.netty-netty-tcnative-boringssl-static-2.0.61.Final-osx-x86_64.jar
-    - io.netty-netty-tcnative-boringssl-static-2.0.61.Final-windows-x86_64.jar
-    - io.netty-netty-tcnative-classes-2.0.61.Final.jar
+    - io.netty-netty-buffer-4.1.108.Final.jar
+    - io.netty-netty-codec-4.1.108.Final.jar
+    - io.netty-netty-codec-dns-4.1.108.Final.jar
+    - io.netty-netty-codec-http-4.1.108.Final.jar
+    - io.netty-netty-codec-http2-4.1.108.Final.jar
+    - io.netty-netty-codec-socks-4.1.108.Final.jar
+    - io.netty-netty-codec-haproxy-4.1.108.Final.jar
+    - io.netty-netty-common-4.1.108.Final.jar
+    - io.netty-netty-handler-4.1.108.Final.jar
+    - io.netty-netty-handler-proxy-4.1.108.Final.jar
+    - io.netty-netty-resolver-4.1.108.Final.jar
+    - io.netty-netty-resolver-dns-4.1.108.Final.jar
+    - io.netty-netty-resolver-dns-classes-macos-4.1.108.Final.jar
+    - io.netty-netty-resolver-dns-native-macos-4.1.108.Final-osx-aarch_64.jar
+    - io.netty-netty-resolver-dns-native-macos-4.1.108.Final-osx-x86_64.jar
+    - io.netty-netty-transport-4.1.108.Final.jar
+    - io.netty-netty-transport-classes-epoll-4.1.108.Final.jar
+    - io.netty-netty-transport-native-epoll-4.1.108.Final-linux-x86_64.jar
+    - io.netty-netty-transport-native-unix-common-4.1.108.Final.jar
+    - io.netty-netty-transport-native-unix-common-4.1.108.Final-linux-x86_64.jar
+    - io.netty-netty-tcnative-boringssl-static-2.0.65.Final.jar
+    - io.netty-netty-tcnative-boringssl-static-2.0.65.Final-linux-aarch_64.jar
+    - io.netty-netty-tcnative-boringssl-static-2.0.65.Final-linux-x86_64.jar
+    - io.netty-netty-tcnative-boringssl-static-2.0.65.Final-osx-aarch_64.jar
+    - io.netty-netty-tcnative-boringssl-static-2.0.65.Final-osx-x86_64.jar
+    - io.netty-netty-tcnative-boringssl-static-2.0.65.Final-windows-x86_64.jar
+    - io.netty-netty-tcnative-classes-2.0.65.Final.jar
     - io.netty.incubator-netty-incubator-transport-classes-io_uring-0.0.21.Final.jar
     - io.netty.incubator-netty-incubator-transport-native-io_uring-0.0.21.Final-linux-x86_64.jar
     - io.netty.incubator-netty-incubator-transport-native-io_uring-0.0.21.Final-linux-aarch_64.jar

distribution/shell/src/assemble/LICENSE.bin.txt

@@ -344,35 +344,35 @@ The Apache Software License, Version 2.0
     - commons-text-1.10.0.jar
     - commons-compress-1.26.0.jar
  * Netty
-    - netty-buffer-4.1.100.Final.jar
-    - netty-codec-4.1.100.Final.jar
-    - netty-codec-dns-4.1.100.Final.jar
-    - netty-codec-http-4.1.100.Final.jar
-    - netty-codec-socks-4.1.100.Final.jar
-    - netty-codec-haproxy-4.1.100.Final.jar
-    - netty-common-4.1.100.Final.jar
-    - netty-handler-4.1.100.Final.jar
-    - netty-handler-proxy-4.1.100.Final.jar
-    - netty-resolver-4.1.100.Final.jar
-    - netty-resolver-dns-4.1.100.Final.jar
-    - netty-transport-4.1.100.Final.jar
-    - netty-transport-classes-epoll-4.1.100.Final.jar
-    - netty-transport-native-epoll-4.1.100.Final-linux-x86_64.jar
-    - netty-transport-native-unix-common-4.1.100.Final.jar
-    - netty-transport-native-unix-common-4.1.100.Final-linux-x86_64.jar
-    - netty-tcnative-boringssl-static-2.0.61.Final.jar
-    - netty-tcnative-boringssl-static-2.0.61.Final-linux-aarch_64.jar
-    - netty-tcnative-boringssl-static-2.0.61.Final-linux-x86_64.jar
-    - netty-tcnative-boringssl-static-2.0.61.Final-osx-aarch_64.jar
-    - netty-tcnative-boringssl-static-2.0.61.Final-osx-x86_64.jar
-    - netty-tcnative-boringssl-static-2.0.61.Final-windows-x86_64.jar
-    - netty-tcnative-classes-2.0.61.Final.jar
+    - netty-buffer-4.1.108.Final.jar
+    - netty-codec-4.1.108.Final.jar
+    - netty-codec-dns-4.1.108.Final.jar
+    - netty-codec-http-4.1.108.Final.jar
+    - netty-codec-socks-4.1.108.Final.jar
+    - netty-codec-haproxy-4.1.108.Final.jar
+    - netty-common-4.1.108.Final.jar
+    - netty-handler-4.1.108.Final.jar
+    - netty-handler-proxy-4.1.108.Final.jar
+    - netty-resolver-4.1.108.Final.jar
+    - netty-resolver-dns-4.1.108.Final.jar
+    - netty-transport-4.1.108.Final.jar
+    - netty-transport-classes-epoll-4.1.108.Final.jar
+    - netty-transport-native-epoll-4.1.108.Final-linux-x86_64.jar
+    - netty-transport-native-unix-common-4.1.108.Final.jar
+    - netty-transport-native-unix-common-4.1.108.Final-linux-x86_64.jar
+    - netty-tcnative-boringssl-static-2.0.65.Final.jar
+    - netty-tcnative-boringssl-static-2.0.65.Final-linux-aarch_64.jar
+    - netty-tcnative-boringssl-static-2.0.65.Final-linux-x86_64.jar
+    - netty-tcnative-boringssl-static-2.0.65.Final-osx-aarch_64.jar
+    - netty-tcnative-boringssl-static-2.0.65.Final-osx-x86_64.jar
+    - netty-tcnative-boringssl-static-2.0.65.Final-windows-x86_64.jar
+    - netty-tcnative-classes-2.0.65.Final.jar
     - netty-incubator-transport-classes-io_uring-0.0.21.Final.jar
     - netty-incubator-transport-native-io_uring-0.0.21.Final-linux-aarch_64.jar
     - netty-incubator-transport-native-io_uring-0.0.21.Final-linux-x86_64.jar
-    - netty-resolver-dns-classes-macos-4.1.100.Final.jar
-    - netty-resolver-dns-native-macos-4.1.100.Final-osx-aarch_64.jar
-    - netty-resolver-dns-native-macos-4.1.100.Final-osx-x86_64.jar
+    - netty-resolver-dns-classes-macos-4.1.108.Final.jar
+    - netty-resolver-dns-native-macos-4.1.108.Final-osx-aarch_64.jar
+    - netty-resolver-dns-native-macos-4.1.108.Final-osx-x86_64.jar
  * Prometheus client
     - simpleclient-0.16.0.jar
     - simpleclient_log4j2-0.16.0.jar

docker/pulsar/Dockerfile

@@ -110,5 +110,5 @@ RUN chmod +x /pulsar/bin/install-pulsar-client.sh
 RUN /pulsar/bin/install-pulsar-client.sh
 
 # The UID must be non-zero. Otherwise, it is arbitrary. No logic should rely on its specific value.
-RUN useradd ${DEFAULT_USERNAME} -u 10000 -g 0
+RUN useradd ${DEFAULT_USERNAME} -u 10000 -g 0 --no-create-home --home-dir /pulsar/data
 USER 10000

docker/pulsar/pom.xml

@@ -47,11 +47,6 @@
     </dependency>
   </dependencies>
 
-  <properties>
-    <UBUNTU_MIRROR>mirror://mirrors.ubuntu.com/mirrors.txt</UBUNTU_MIRROR>
-    <UBUNTU_SECURITY_MIRROR>http://security.ubuntu.com/ubuntu/</UBUNTU_SECURITY_MIRROR>
-  </properties>
-
   <profiles>
     <profile>
       <id>git-commit-id-no-git</id>

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java

@@ -350,15 +350,19 @@ private CompletableFuture<Void> computeCursorProperties(
             final Function<Map<String, String>, Map<String, String>> updateFunction) {
         CompletableFuture<Void> updateCursorPropertiesResult = new CompletableFuture<>();
 
-        final Stat lastCursorLedgerStat = ManagedCursorImpl.this.cursorLedgerStat;
-
         Map<String, String> newProperties = updateFunction.apply(ManagedCursorImpl.this.cursorProperties);
+        if (!isDurable()) {
+            this.cursorProperties = Collections.unmodifiableMap(newProperties);
+            updateCursorPropertiesResult.complete(null);
+            return updateCursorPropertiesResult;
+        }
+
         ManagedCursorInfo copy = ManagedCursorInfo
                 .newBuilder(ManagedCursorImpl.this.managedCursorInfo)
                 .clearCursorProperties()
                 .addAllCursorProperties(buildStringPropertiesMap(newProperties))
                 .build();
-
+        final Stat lastCursorLedgerStat = ManagedCursorImpl.this.cursorLedgerStat;
         ledger.getStore().asyncUpdateCursorInfo(ledger.getName(),
                 name, copy, lastCursorLedgerStat, new MetaStoreCallback<>() {
                     @Override

managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java

@@ -2440,7 +2440,7 @@ public void testRetentionSize() throws Exception {
         });
     }
 
-    @Test
+    @Test(groups = "flaky")
     public void testTimestampOnWorkingLedger() throws Exception {
         ManagedLedgerConfig conf = new ManagedLedgerConfig();
         conf.setMaxEntriesPerLedger(1);
@@ -3505,7 +3505,7 @@ public void testLedgerReachMaximumRolloverTime() throws Exception {
                 .until(() -> firstLedgerId != ml.addEntry("test".getBytes()).getLedgerId());
     }
 
-    @Test
+    @Test(groups = "flaky")
     public void testLedgerNotRolloverWithoutOpenState() throws Exception {
         ManagedLedgerConfig config = new ManagedLedgerConfig();
         config.setMaxEntriesPerLedger(2);

pom.xml

@@ -84,11 +84,14 @@ flexible messaging model and an intuitive client API.</description>
 
     <pulsar.client.python.version>3.4.0</pulsar.client.python.version>
 
+    <UBUNTU_MIRROR>http://archive.ubuntu.com/ubuntu/</UBUNTU_MIRROR>
+    <UBUNTU_SECURITY_MIRROR>http://security.ubuntu.com/ubuntu/</UBUNTU_SECURITY_MIRROR>
+
     <!--config keys to configure test selection -->
     <include>**/Test*.java,**/*Test.java,**/*Tests.java,**/*TestCase.java</include>
     <exclude/>
     <groups/>
-    <excludedGroups>quarantine</excludedGroups>
+    <excludedGroups>quarantine,flaky</excludedGroups>
 
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
@@ -140,7 +143,7 @@ flexible messaging model and an intuitive client API.</description>
     <snappy.version>1.1.10.5</snappy.version> <!-- ZooKeeper server -->
     <dropwizardmetrics.version>4.1.12.1</dropwizardmetrics.version> <!-- ZooKeeper server -->
     <curator.version>5.1.0</curator.version>
-    <netty.version>4.1.100.Final</netty.version>
+    <netty.version>4.1.108.Final</netty.version>
     <netty-iouring.version>0.0.21.Final</netty-iouring.version>
     <jetty.version>9.4.54.v20240208</jetty.version>
     <conscrypt.version>2.5.2</conscrypt.version>
@@ -1613,6 +1616,9 @@ flexible messaging model and an intuitive client API.</description>
               <value>org.apache.pulsar.tests.PulsarTestListener,org.apache.pulsar.tests.JacocoDumpListener,org.apache.pulsar.tests.AnnotationListener,org.apache.pulsar.tests.FailFastNotifier,org.apache.pulsar.tests.MockitoCleanupListener,org.apache.pulsar.tests.FastThreadLocalCleanupListener,org.apache.pulsar.tests.ThreadLeakDetectorListener,org.apache.pulsar.tests.SingletonCleanerListener</value>
             </property>
           </properties>
+          <excludes>
+            <exclude>**/AlluxioSink*.java</exclude>
+          </excludes>
         </configuration>
         <dependencies>
           <dependency>
@@ -2691,6 +2697,32 @@ flexible messaging model and an intuitive client API.</description>
         </repository>
       </distributionManagement>
     </profile>
+
+    <profile>
+      <id>ubuntu-mirror-set</id>
+      <activation>
+        <property>
+          <name>env.UBUNTU_MIRROR</name>
+        </property>
+      </activation>
+      <properties>
+        <!-- Override the default value with the environment variable -->
+        <UBUNTU_MIRROR>${env.UBUNTU_MIRROR}</UBUNTU_MIRROR>
+      </properties>
+    </profile>
+    <profile>
+      <id>ubuntu-security-mirror-set</id>
+      <activation>
+        <property>
+          <name>env.UBUNTU_SECURITY_MIRROR</name>
+        </property>
+      </activation>
+      <properties>
+        <!-- Override the default value with the environment variable -->
+        <UBUNTU_SECURITY_MIRROR>${env.UBUNTU_SECURITY_MIRROR}</UBUNTU_SECURITY_MIRROR>
+      </properties>
+    </profile>
+
   </profiles>
 
   <repositories>

pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java

@@ -379,7 +379,7 @@ public void closeMetadataServiceSession() throws Exception {
     }
 
     private void closeLeaderElectionService() throws Exception {
-        if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(config)) {
+        if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(this)) {
             ExtensibleLoadManagerImpl.get(loadManager.get()).getLeaderElectionService().close();
         } else {
             if (this.leaderElectionService != null) {
@@ -1135,7 +1135,7 @@ protected void closeLocalMetadataStore() throws Exception {
     }
 
     protected void startLeaderElectionService() {
-        if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(config)) {
+        if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(this)) {
             LOG.info("The load manager extension is enabled. Skipping PulsarService LeaderElectionService.");
             return;
         }
@@ -1250,7 +1250,7 @@ protected void startLoadManagementService() throws PulsarServerException {
         LOG.info("Starting load management service ...");
         this.loadManager.get().start();
 
-        if (config.isLoadBalancerEnabled() && !ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(config)) {
+        if (config.isLoadBalancerEnabled() && !ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(this)) {
             LOG.info("Starting load balancer");
             if (this.loadReportTask == null) {
                 long loadReportMinInterval = config.getLoadBalancerReportUpdateMinIntervalMillis();
@@ -1343,7 +1343,7 @@ public boolean isRunning() {
      * @return a reference of the current <code>LeaderElectionService</code> instance.
      */
     public LeaderElectionService getLeaderElectionService() {
-        if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(config)) {
+        if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(this)) {
             return ExtensibleLoadManagerImpl.get(loadManager.get()).getLeaderElectionService();
         } else {
             return this.leaderElectionService;

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java

@@ -968,13 +968,13 @@ public CompletableFuture<Void> setNamespaceBundleAffinityAsync(String bundleRang
                     return CompletableFuture.completedFuture(null);
                 })
                 .thenCompose(__ -> {
-                    if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(config())) {
+                    if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar())) {
                         return CompletableFuture.completedFuture(null);
                     }
                     return validateLeaderBrokerAsync();
                 })
                 .thenAccept(__ -> {
-                    if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(config())) {
+                    if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar())) {
                         return;
                     }
                     // For ExtensibleLoadManager, this operation will be ignored.