From 6556f43e3fc47b395ddbea1b196d755b5d946c18 Mon Sep 17 00:00:00 2001 From: Heesung Sohn <103456639+heesung-sn@users.noreply.github.com> Date: Wed, 3 Apr 2024 21:01:49 -0700 Subject: [PATCH 1/7] [fix][broker] Update TransferShedder underloaded broker check to consider max loaded broker's msgThroughputEMA and update IsExtensibleLoadBalancerImpl check (#22321) (#22417) (cherry picked from commit 651908a903301da9c07dc93300635cc28d8ee69f) --- .../apache/pulsar/broker/PulsarService.java | 8 ++--- .../broker/admin/impl/NamespacesBase.java | 4 +-- .../extensions/ExtensibleLoadManagerImpl.java | 4 --- .../extensions/scheduler/TransferShedder.java | 22 +++++++++----- .../broker/namespace/NamespaceService.java | 30 +++++++++---------- .../pulsar/broker/web/PulsarWebResource.java | 4 +-- .../ExtensibleLoadManagerImplTest.java | 1 - .../scheduler/TransferShedderTest.java | 13 ++++---- 8 files changed, 45 insertions(+), 41 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java index 64f4ee0288122..5410bacbe78eb 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java +++ b/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 LeaderElectionService instance. */ public LeaderElectionService getLeaderElectionService() { - if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(config)) { + if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(this)) { return ExtensibleLoadManagerImpl.get(loadManager.get()).getLeaderElectionService(); } else { return this.leaderElectionService; diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java index 9478857032f22..f4732cad38040 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java @@ -968,13 +968,13 @@ public CompletableFuture 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. diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java index 6a0e677c66268..26ee45b7444f3 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java @@ -263,10 +263,6 @@ public ExtensibleLoadManagerImpl() { this.brokerSelectionStrategy = new LeastResourceUsageWithWeight(); } - public static boolean isLoadManagerExtensionEnabled(ServiceConfiguration conf) { - return ExtensibleLoadManagerImpl.class.getName().equals(conf.getLoadManagerClassName()); - } - public static boolean isLoadManagerExtensionEnabled(PulsarService pulsar) { return pulsar.getLoadManager().get() instanceof ExtensibleLoadManagerWrapper; } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/TransferShedder.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/TransferShedder.java index 3564b4e9e3b94..7126ccb034196 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/TransferShedder.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/TransferShedder.java @@ -362,7 +362,7 @@ public Set findBundlesForUnloading(LoadManagerContext context, final double targetStd = conf.getLoadBalancerBrokerLoadTargetStd(); boolean transfer = conf.isLoadBalancerTransferEnabled(); if (stats.std() > targetStd - || isUnderLoaded(context, stats.peekMinBroker(), stats.avg) + || isUnderLoaded(context, stats.peekMinBroker(), stats) || isOverLoaded(context, stats.peekMaxBroker(), stats.avg)) { unloadConditionHitCount++; } else { @@ -390,7 +390,7 @@ public Set findBundlesForUnloading(LoadManagerContext context, UnloadDecision.Reason reason; if (stats.std() > targetStd) { reason = Overloaded; - } else if (isUnderLoaded(context, stats.peekMinBroker(), stats.avg)) { + } else if (isUnderLoaded(context, stats.peekMinBroker(), stats)) { reason = Underloaded; if (debugMode) { log.info(String.format("broker:%s is underloaded:%s although " @@ -669,19 +669,27 @@ public Set findBundlesForUnloading(LoadManagerContext context, } - private boolean isUnderLoaded(LoadManagerContext context, String broker, double avgLoad) { + private boolean isUnderLoaded(LoadManagerContext context, String broker, LoadStats stats) { var brokerLoadDataOptional = context.brokerLoadDataStore().get(broker); if (brokerLoadDataOptional.isEmpty()) { return false; } var brokerLoadData = brokerLoadDataOptional.get(); - if (brokerLoadData.getMsgThroughputEMA() < 1) { + + var underLoadedMultiplier = + Math.min(0.5, Math.max(0.0, context.brokerConfiguration().getLoadBalancerBrokerLoadTargetStd() / 2.0)); + + if (brokerLoadData.getWeightedMaxEMA() < stats.avg * underLoadedMultiplier) { return true; } - return brokerLoadData.getWeightedMaxEMA() - < avgLoad * Math.min(0.5, Math.max(0.0, - context.brokerConfiguration().getLoadBalancerBrokerLoadTargetStd() / 2)); + var maxBrokerLoadDataOptional = context.brokerLoadDataStore().get(stats.peekMaxBroker()); + if (maxBrokerLoadDataOptional.isEmpty()) { + return false; + } + + return brokerLoadData.getMsgThroughputEMA() + < maxBrokerLoadDataOptional.get().getMsgThroughputEMA() * underLoadedMultiplier; } private boolean isOverLoaded(LoadManagerContext context, String broker, double avgLoad) { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java index 61e045ed304fd..e04be25fe499c 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java @@ -195,7 +195,7 @@ public CompletableFuture> getBrokerServiceUrlAsync(TopicN pulsar.getBrokerId(), optResult.get(), topic); return CompletableFuture.completedFuture(optResult); } - if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(config)) { + if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar)) { return loadManager.get().findBrokerServiceUrl(Optional.of(topic), bundle); } else { // TODO: Add unit tests cover it. @@ -311,7 +311,7 @@ private CompletableFuture> internalGetWebServiceUrl(@Nullable Serv return CompletableFuture.completedFuture(Optional.empty()); } CompletableFuture> future = - ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(config) + ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar) ? loadManager.get().findBrokerServiceUrl(Optional.ofNullable(topic), bundle) : findBrokerServiceUrl(bundle, options); @@ -375,7 +375,7 @@ public boolean registerNamespace(NamespaceName nsname, boolean ensureOwned) thro NamespaceBundle nsFullBundle = bundleFactory.getFullBundle(nsname); // v2 namespace will always use full bundle object final NamespaceEphemeralData otherData; - if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(config)) { + if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar)) { ExtensibleLoadManagerImpl loadManager = ExtensibleLoadManagerImpl.get(this.loadManager.get()); otherData = loadManager.tryAcquiringOwnership(nsFullBundle).get(); } else { @@ -781,7 +781,7 @@ public CompletableFuture unloadNamespaceBundle(NamespaceBundle bundle, long timeout, TimeUnit timeoutUnit, boolean closeWithoutWaitingClientDisconnect) { - if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(config)) { + if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar)) { return ExtensibleLoadManagerImpl.get(loadManager.get()) .unloadNamespaceBundleAsync(bundle, destinationBroker); } @@ -803,7 +803,7 @@ public CompletableFuture> getOwnedNameSpac .getIsolationDataPoliciesAsync(pulsar.getConfiguration().getClusterName()) .thenApply(nsIsolationPoliciesOpt -> nsIsolationPoliciesOpt.orElseGet(NamespaceIsolationPolicies::new)) .thenCompose(namespaceIsolationPolicies -> { - if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(config)) { + if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar)) { ExtensibleLoadManagerImpl extensibleLoadManager = ExtensibleLoadManagerImpl.get(loadManager.get()); var statusMap = extensibleLoadManager.getOwnedServiceUnits().stream() @@ -883,7 +883,7 @@ public boolean isNamespaceBundleDisabled(NamespaceBundle bundle) throws Exceptio public CompletableFuture splitAndOwnBundle(NamespaceBundle bundle, boolean unload, NamespaceBundleSplitAlgorithm splitAlgorithm, List boundaries) { - if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(config)) { + if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar)) { return ExtensibleLoadManagerImpl.get(loadManager.get()) .splitNamespaceBundleAsync(bundle, splitAlgorithm, boundaries); } @@ -1126,7 +1126,7 @@ public OwnershipCache getOwnershipCache() { } public Set getOwnedServiceUnits() { - if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(config)) { + if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar)) { ExtensibleLoadManagerImpl extensibleLoadManager = ExtensibleLoadManagerImpl.get(loadManager.get()); return extensibleLoadManager.getOwnedServiceUnits(); } @@ -1148,7 +1148,7 @@ public CompletableFuture isServiceUnitOwnedAsync(ServiceUnitId suName) } if (suName instanceof NamespaceBundle) { - if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(config)) { + if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar)) { return loadManager.get().checkOwnershipAsync(Optional.empty(), suName); } // TODO: Add unit tests cover it. @@ -1176,7 +1176,7 @@ public boolean isServiceUnitActive(TopicName topicName) { public CompletableFuture isServiceUnitActiveAsync(TopicName topicName) { // TODO: Add unit tests cover it. - if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(config)) { + if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar)) { return getBundleAsync(topicName) .thenCompose(bundle -> loadManager.get().checkOwnershipAsync(Optional.of(topicName), bundle)); } @@ -1191,7 +1191,7 @@ public CompletableFuture isServiceUnitActiveAsync(TopicName topicName) private CompletableFuture isNamespaceOwnedAsync(NamespaceName fqnn) { // TODO: Add unit tests cover it. - if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(config)) { + if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar)) { return getFullBundleAsync(fqnn) .thenCompose(bundle -> loadManager.get().checkOwnershipAsync(Optional.empty(), bundle)); } @@ -1201,7 +1201,7 @@ private CompletableFuture isNamespaceOwnedAsync(NamespaceName fqnn) { private CompletableFuture isTopicOwnedAsync(TopicName topic) { // TODO: Add unit tests cover it. - if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(config)) { + if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar)) { return getBundleAsync(topic) .thenCompose(bundle -> loadManager.get().checkOwnershipAsync(Optional.of(topic), bundle)); } @@ -1210,7 +1210,7 @@ private CompletableFuture isTopicOwnedAsync(TopicName topic) { public CompletableFuture checkTopicOwnership(TopicName topicName) { // TODO: Add unit tests cover it. - if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(config)) { + if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar)) { return getBundleAsync(topicName) .thenCompose(bundle -> loadManager.get().checkOwnershipAsync(Optional.of(topicName), bundle)); } @@ -1220,7 +1220,7 @@ public CompletableFuture checkTopicOwnership(TopicName topicName) { public CompletableFuture removeOwnedServiceUnitAsync(NamespaceBundle nsBundle) { CompletableFuture future; - if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(config)) { + if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar)) { ExtensibleLoadManagerImpl extensibleLoadManager = ExtensibleLoadManagerImpl.get(loadManager.get()); future = extensibleLoadManager.unloadNamespaceBundleAsync(nsBundle, Optional.empty()); } else { @@ -1528,7 +1528,7 @@ public PulsarClientImpl getNamespaceClient(ClusterDataImpl cluster) { } public CompletableFuture> getOwnerAsync(NamespaceBundle bundle) { - if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(config)) { + if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar)) { ExtensibleLoadManagerImpl extensibleLoadManager = ExtensibleLoadManagerImpl.get(loadManager.get()); return extensibleLoadManager.getOwnershipWithLookupDataAsync(bundle) .thenCompose(lookupData -> lookupData @@ -1545,7 +1545,7 @@ public boolean checkOwnershipPresent(NamespaceBundle bundle) throws Exception { } public CompletableFuture checkOwnershipPresentAsync(NamespaceBundle bundle) { - if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(config)) { + if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar)) { ExtensibleLoadManagerImpl extensibleLoadManager = ExtensibleLoadManagerImpl.get(loadManager.get()); return extensibleLoadManager.getOwnershipAsync(Optional.empty(), bundle) .thenApply(Optional::isPresent); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java index 0d01a2c50419d..2f437962002a3 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java @@ -732,7 +732,7 @@ public CompletableFuture validateBundleOwnershipAsync(NamespaceBundle bund .host(webUrl.get().getHost()) .port(webUrl.get().getPort()) .replaceQueryParam("authoritative", newAuthoritative); - if (!ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(config())) { + if (!ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar)) { uriBuilder.replaceQueryParam("destinationBroker", null); } URI redirect = uriBuilder.build(); @@ -1007,7 +1007,7 @@ protected boolean isLeaderBroker() { protected static boolean isLeaderBroker(PulsarService pulsar) { // For extensible load manager, it doesn't have leader election service on pulsar broker. - if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar.getConfig())) { + if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar)) { return true; } return pulsar.getLeaderElectionService().isLeader(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java index 9c363325c4cc8..03ea937c44edc 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java @@ -317,7 +317,6 @@ public boolean test(NamespaceBundle namespaceBundle) { assertEquals(unloadCount.get(), 1); }); - String dstBrokerUrl = pulsar1.getBrokerId(); String dstBrokerServiceUrl; if (broker.equals(pulsar1.getBrokerServiceUrl())) { diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/TransferShedderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/TransferShedderTest.java index 0ff64616973d9..efca2880949f2 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/TransferShedderTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/TransferShedderTest.java @@ -1104,16 +1104,17 @@ public void testUnloadBundlesGreaterThanTargetThroughputAfterSplit() throws Ille assertEquals(stats.std(), 2.5809568279517847E-8); } - @Test - public void testMinBrokerWithZeroTraffic() throws IllegalAccessException { + public void testMinBrokerWithLowTraffic() throws IllegalAccessException { UnloadCounter counter = new UnloadCounter(); TransferShedder transferShedder = new TransferShedder(counter); var ctx = setupContext(); var brokerLoadDataStore = ctx.brokerLoadDataStore(); - var load = getCpuLoad(ctx, 4, "broker2:8080"); - FieldUtils.writeDeclaredField(load,"msgThroughputEMA", 0, true); + var load = getCpuLoad(ctx, 4, "broker2:8080"); + FieldUtils.writeDeclaredField(load, "msgThroughputEMA", 10, true); + + brokerLoadDataStore.pushAsync("broker2:8080", load); brokerLoadDataStore.pushAsync("broker4:8080", getCpuLoad(ctx, 55, "broker4:8080")); brokerLoadDataStore.pushAsync("broker5:8080", getCpuLoad(ctx, 65, "broker5:8080")); @@ -1268,10 +1269,10 @@ public void testOverloadOutlier() { Assertions.assertThat(res).isIn( Set.of(new UnloadDecision( new Unload("broker99:8080", "my-tenant/my-namespace99/0x00000000_0x0FFFFFFF", - Optional.of("broker52:8080")), Success, Overloaded)), + Optional.of("broker52:8080")), Success, Underloaded)), Set.of(new UnloadDecision( new Unload("broker99:8080", "my-tenant/my-namespace99/0x00000000_0x0FFFFFFF", - Optional.of("broker83:8080")), Success, Overloaded)) + Optional.of("broker83:8080")), Success, Underloaded)) ); assertEquals(counter.getLoadAvg(), 0.019900000000000008, 0.00001); assertEquals(counter.getLoadStd(), 0.09850375627355534, 0.00001); From 0409027e5dda05462a7c21802439b0b6609c0bb7 Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Wed, 27 Mar 2024 12:34:14 -0700 Subject: [PATCH 2/7] [improve][misc] Upgrade to Netty 4.1.108 and tcnative 2.0.65 (#22369) (cherry picked from commit be0a9d9d9bb23dabc065f091b853f27c0ebcaa16) (cherry picked from commit 1987398a19df03743f636ae7493ae92cc8cd2e82) --- buildtools/pom.xml | 2 +- .../server/src/assemble/LICENSE.bin.txt | 54 +++++++++---------- .../shell/src/assemble/LICENSE.bin.txt | 52 +++++++++--------- pom.xml | 2 +- pulsar-sql/presto-distribution/LICENSE | 54 +++++++++---------- 5 files changed, 82 insertions(+), 82 deletions(-) diff --git a/buildtools/pom.xml b/buildtools/pom.xml index 56002fe068bef..61c56f65a2ab3 100644 --- a/buildtools/pom.xml +++ b/buildtools/pom.xml @@ -47,7 +47,7 @@ 4.1 10.14.2 3.1.2 - 4.1.100.Final + 4.1.108.Final 4.2.3 32.1.1-jre 1.10.12 diff --git a/distribution/server/src/assemble/LICENSE.bin.txt b/distribution/server/src/assemble/LICENSE.bin.txt index b7da9969a4bf8..d42c7cc9fe1bd 100644 --- a/distribution/server/src/assemble/LICENSE.bin.txt +++ b/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 diff --git a/distribution/shell/src/assemble/LICENSE.bin.txt b/distribution/shell/src/assemble/LICENSE.bin.txt index 5d880a018c53f..df04985a3a880 100644 --- a/distribution/shell/src/assemble/LICENSE.bin.txt +++ b/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 diff --git a/pom.xml b/pom.xml index ffa5f1178ad5e..33e979083ffe0 100644 --- a/pom.xml +++ b/pom.xml @@ -140,7 +140,7 @@ flexible messaging model and an intuitive client API. 1.1.10.5 4.1.12.1 5.1.0 - 4.1.100.Final + 4.1.108.Final 0.0.21.Final 9.4.54.v20240208 2.5.2 diff --git a/pulsar-sql/presto-distribution/LICENSE b/pulsar-sql/presto-distribution/LICENSE index df7887fed3e59..079b033c0942b 100644 --- a/pulsar-sql/presto-distribution/LICENSE +++ b/pulsar-sql/presto-distribution/LICENSE @@ -231,34 +231,34 @@ The Apache Software License, Version 2.0 - commons-compress-1.26.0.jar - commons-lang3-3.11.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-haproxy-4.1.100.Final.jar - - netty-codec-socks-4.1.100.Final.jar - - netty-handler-proxy-4.1.100.Final.jar - - netty-common-4.1.100.Final.jar - - netty-handler-4.1.100.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-haproxy-4.1.108.Final.jar + - netty-codec-socks-4.1.108.Final.jar + - netty-handler-proxy-4.1.108.Final.jar + - netty-common-4.1.108.Final.jar + - netty-handler-4.1.108.Final.jar - netty-reactive-streams-2.0.6.jar - - netty-resolver-4.1.100.Final.jar - - netty-resolver-dns-4.1.100.Final.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-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-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-codec-http2-4.1.100.Final.jar + - netty-resolver-4.1.108.Final.jar + - netty-resolver-dns-4.1.108.Final.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 + - 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-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-codec-http2-4.1.108.Final.jar - netty-incubator-transport-classes-io_uring-0.0.21.Final.jar - netty-incubator-transport-native-io_uring-0.0.21.Final-linux-x86_64.jar - netty-incubator-transport-native-io_uring-0.0.21.Final-linux-aarch_64.jar From 2283dc160f75dbc314c5f5abcba44455282f3981 Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Thu, 4 Apr 2024 11:25:16 -0700 Subject: [PATCH 3/7] [improve][test] Move most flaky tests to flaky group (#22433) - also add solution for running test methods added to flaky group since that was missing (cherry picked from commit 5f31ec383bb7526eca24b95002f6cd498057fee7) # Conflicts: # pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicAuthZTest.java # pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java (cherry picked from commit 097805d68eb45504da31e1cd0444cea4147c8d4b) --- build/run_unit_group.sh | 14 +++++++++--- .../pulsar/tests/AnnotationListener.java | 22 +++++++++++++++++++ .../mledger/impl/ManagedLedgerTest.java | 4 ++-- pom.xml | 2 +- .../admin/AdminApiMultiBrokersTest.java | 2 +- .../pulsar/broker/admin/TopicAuthZTest.java | 4 ++-- .../service/PersistentMessageFinderTest.java | 6 ++--- .../impl/ProducerConsumerInternalTest.java | 2 +- .../client/impl/TransactionEndToEndTest.java | 4 ++-- ...ctionEndToEndWithoutBatchIndexAckTest.java | 2 +- 10 files changed, 46 insertions(+), 16 deletions(-) diff --git a/build/run_unit_group.sh b/build/run_unit_group.sh index 8597848b69bf8..b6275903be6f0 100755 --- a/build/run_unit_group.sh +++ b/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::" diff --git a/buildtools/src/main/java/org/apache/pulsar/tests/AnnotationListener.java b/buildtools/src/main/java/org/apache/pulsar/tests/AnnotationListener.java index 38cd2a1747a63..0c464fd97a970 100644 --- a/buildtools/src/main/java/org/apache/pulsar/tests/AnnotationListener.java +++ b/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) { diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java index 34f65dfd00ee8..94692aeea976a 100644 --- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java +++ b/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); diff --git a/pom.xml b/pom.xml index 33e979083ffe0..27243281bd98a 100644 --- a/pom.xml +++ b/pom.xml @@ -88,7 +88,7 @@ flexible messaging model and an intuitive client API. **/Test*.java,**/*Test.java,**/*Tests.java,**/*TestCase.java - quarantine + quarantine,flaky UTF-8 UTF-8 diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiMultiBrokersTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiMultiBrokersTest.java index 7c9154a27ff69..46b24abd6d4e2 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiMultiBrokersTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiMultiBrokersTest.java @@ -132,7 +132,7 @@ public void testTopicLookup(TopicDomain topicDomain, boolean isPartition) throws Assert.assertEquals(lookupResultSet.size(), 1); } - @Test + @Test(groups = "flaky") public void testForceDeletePartitionedTopicWithSub() throws Exception { final int numPartitions = 10; TenantInfoImpl tenantInfo = new TenantInfoImpl(Set.of("role1", "role2"), Set.of("test")); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicAuthZTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicAuthZTest.java index e23f9bbaf9b30..9e975ba0854d4 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicAuthZTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicAuthZTest.java @@ -53,7 +53,7 @@ public class TopicAuthZTest extends MockedPulsarStandalone { .claim("sub", TENANT_ADMIN_SUBJECT).signWith(SECRET_KEY).compact(); @SneakyThrows - @BeforeClass + @BeforeClass(alwaysRun = true) public void before() { configureTokenAuthentication(); configureDefaultAuthorization(); @@ -73,7 +73,7 @@ public void before() { @SneakyThrows - @AfterClass + @AfterClass(alwaysRun = true) public void after() { if (superUserAdmin != null) { superUserAdmin.close(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentMessageFinderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentMessageFinderTest.java index ace552a55a72a..ac5ab94c213c5 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentMessageFinderTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentMessageFinderTest.java @@ -266,7 +266,7 @@ void testPersistentMessageFinderWhenLastMessageDelete() throws Exception { ledger.addEntry(createMessageWrittenToLedger("msg2")); ledger.addEntry(createMessageWrittenToLedger("msg3")); Position lastPosition = ledger.addEntry(createMessageWrittenToLedger("last-message")); - + long endTimestamp = System.currentTimeMillis() + 1000; Result result = new Result(); @@ -383,7 +383,7 @@ public static Set getBrokerEntryMetadataIntercep * * @throws Exception */ - @Test + @Test(groups = "flaky") void testMessageExpiryWithTimestampNonRecoverableException() throws Exception { final String ledgerAndCursorName = "testPersistentMessageExpiryWithNonRecoverableLedgers"; @@ -440,7 +440,7 @@ void testMessageExpiryWithTimestampNonRecoverableException() throws Exception { } - @Test + @Test(groups = "flaky") public void testIncorrectClientClock() throws Exception { final String ledgerAndCursorName = "testIncorrectClientClock"; int maxTTLSeconds = 1; diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerConsumerInternalTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerConsumerInternalTest.java index 240d8d2304768..4ec8107030600 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerConsumerInternalTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerConsumerInternalTest.java @@ -116,7 +116,7 @@ private void removeServiceProducerMaintainedByServerCnx(ServiceProducer serviceP }); } - @Test + @Test(groups = "flaky") public void testExclusiveConsumerWillAlwaysRetryEvenIfReceivedConsumerBusyError() throws Exception { final String topicName = BrokerTestUtil.newUniqueName("persistent://my-property/my-ns/tp_"); final String subscriptionName = "subscription1"; diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java index 348fb04b7dd23..4abcb09c0fe6d 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java @@ -96,7 +96,7 @@ public class TransactionEndToEndTest extends TransactionTestBase { protected static final String TOPIC_OUTPUT = NAMESPACE1 + "/output"; protected static final String TOPIC_MESSAGE_ACK_TEST = NAMESPACE1 + "/message-ack-test"; protected static final int NUM_PARTITIONS = 16; - @BeforeClass + @BeforeClass(alwaysRun = true) protected void setup() throws Exception { conf.setAcknowledgmentAtBatchIndexLevelEnabled(true); setUpBase(1, NUM_PARTITIONS, TOPIC_OUTPUT, TOPIC_PARTITION); @@ -1624,7 +1624,7 @@ public void testSendTxnAckBatchMessageToDLQ() throws Exception { admin.topics().delete(topic, true); } - @Test + @Test(groups = "flaky") public void testDelayedTransactionMessages() throws Exception { String topic = NAMESPACE1 + "/testDelayedTransactionMessages"; diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndWithoutBatchIndexAckTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndWithoutBatchIndexAckTest.java index 52faae2f8ea1f..df4ad32b6a8ae 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndWithoutBatchIndexAckTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndWithoutBatchIndexAckTest.java @@ -30,7 +30,7 @@ @Test(groups = "broker-impl") public class TransactionEndToEndWithoutBatchIndexAckTest extends TransactionEndToEndTest { - @BeforeClass + @BeforeClass(alwaysRun = true) protected void setup() throws Exception { conf.setAcknowledgmentAtBatchIndexLevelEnabled(false); setUpBase(1, NUM_PARTITIONS, TOPIC_OUTPUT, TOPIC_PARTITION); From 1b52293705d6839841796bd3723baa158cb0d59c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=81=93=E5=90=9B?= Date: Fri, 5 Apr 2024 15:42:40 +0800 Subject: [PATCH 4/7] [fix][broker][admin] Fix cannot update properties on NonDurable subscription. (#22411) (cherry picked from commit 902728ef6590233b87c14d2528590ad7e6fdcc12) # Conflicts: # pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java (cherry picked from commit 7cd0924bc9996f95b36601958ae156408614c523) --- .../mledger/impl/ManagedCursorImpl.java | 10 +++-- .../broker/admin/PersistentTopicsTest.java | 38 +++++++++++++++++++ 2 files changed, 45 insertions(+), 3 deletions(-) diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java index 8db8a5714398c..f67f534f86d45 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 @@ -350,15 +350,19 @@ private CompletableFuture computeCursorProperties( final Function, Map> updateFunction) { CompletableFuture updateCursorPropertiesResult = new CompletableFuture<>(); - final Stat lastCursorLedgerStat = ManagedCursorImpl.this.cursorLedgerStat; - Map 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 diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java index 1f755234009ba..c7b066faea9da 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java @@ -32,6 +32,7 @@ import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertNotNull; import static org.testng.Assert.assertSame; import static org.testng.Assert.assertTrue; import java.lang.reflect.Field; @@ -53,6 +54,8 @@ import javax.ws.rs.core.UriInfo; import lombok.Cleanup; import lombok.extern.slf4j.Slf4j; +import org.apache.bookkeeper.mledger.ManagedCursor; +import org.apache.commons.collections4.MapUtils; import org.apache.pulsar.broker.BrokerTestUtil; import org.apache.pulsar.broker.admin.v2.ExtPersistentTopics; import org.apache.pulsar.broker.admin.v2.NonPersistentTopics; @@ -65,6 +68,8 @@ import org.apache.pulsar.broker.resources.TopicResources; import org.apache.pulsar.broker.service.BrokerService; import org.apache.pulsar.broker.service.Topic; +import org.apache.pulsar.broker.service.persistent.PersistentSubscription; +import org.apache.pulsar.broker.service.persistent.PersistentTopic; import org.apache.pulsar.broker.web.PulsarWebResource; import org.apache.pulsar.broker.web.RestException; import org.apache.pulsar.client.admin.LongRunningProcessStatus; @@ -76,6 +81,7 @@ import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.MessageId; import org.apache.pulsar.client.api.Producer; +import org.apache.pulsar.client.api.Reader; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.api.interceptor.ProducerInterceptor; import org.apache.pulsar.client.impl.BatchMessageIdImpl; @@ -1773,4 +1779,36 @@ public void testNamespaceResources() throws Exception { assertTrue(namespaces.contains(ns1V2)); assertTrue(namespaces.contains(ns1V1)); } + + @Test + public void testUpdatePropertiesOnNonDurableSub() throws Exception { + String topic = "persistent://" + testTenant + "/" + testNamespaceLocal + "/testUpdatePropertiesOnNonDurableSub"; + String subscription = "sub"; + admin.topics().createNonPartitionedTopic(topic); + + @Cleanup + Reader __ = pulsarClient.newReader(Schema.STRING) + .startMessageId(MessageId.earliest) + .subscriptionName(subscription) + .topic(topic) + .create(); + + PersistentTopic persistentTopic = + (PersistentTopic) pulsar.getBrokerService().getTopic(topic, false).get().get(); + PersistentSubscription subscription1 = persistentTopic.getSubscriptions().get(subscription); + assertNotNull(subscription1); + ManagedCursor cursor = subscription1.getCursor(); + + Map properties = admin.topics().getSubscriptionProperties(topic, subscription); + assertEquals(properties.size(), 0); + assertTrue(MapUtils.isEmpty(cursor.getCursorProperties())); + + admin.topics().updateSubscriptionProperties(topic, subscription, Map.of("foo", "bar")); + properties = admin.topics().getSubscriptionProperties(topic, subscription); + assertEquals(properties.size(), 1); + assertEquals(properties.get("foo"), "bar"); + + assertEquals(cursor.getCursorProperties().size(), 1); + assertEquals(cursor.getCursorProperties().get("foo"), "bar"); + } } From 95d8662d28c33b6bfd033036b571cb2568a2de25 Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Thu, 22 Feb 2024 14:42:26 +0200 Subject: [PATCH 5/7] [fix][build] Fix building docker images without setting UBUNTU_MIRROR - similar change was made in master as part of #22095 (cherry picked from commit c27bcf000322720601f2ab8308684fed703ce546) Conflicts: pom.xml --- docker/pulsar/pom.xml | 5 ---- pom.xml | 29 +++++++++++++++++++++ tests/docker-images/java-test-image/pom.xml | 4 +-- 3 files changed, 31 insertions(+), 7 deletions(-) diff --git a/docker/pulsar/pom.xml b/docker/pulsar/pom.xml index 9873f2c6bec06..eab6428939700 100644 --- a/docker/pulsar/pom.xml +++ b/docker/pulsar/pom.xml @@ -47,11 +47,6 @@ - - mirror://mirrors.ubuntu.com/mirrors.txt - http://security.ubuntu.com/ubuntu/ - - git-commit-id-no-git diff --git a/pom.xml b/pom.xml index 27243281bd98a..ccb127f2d1066 100644 --- a/pom.xml +++ b/pom.xml @@ -84,6 +84,9 @@ flexible messaging model and an intuitive client API. 3.4.0 + http://archive.ubuntu.com/ubuntu/ + http://security.ubuntu.com/ubuntu/ + **/Test*.java,**/*Test.java,**/*Tests.java,**/*TestCase.java @@ -2691,6 +2694,32 @@ flexible messaging model and an intuitive client API. + + + ubuntu-mirror-set + + + env.UBUNTU_MIRROR + + + + + ${env.UBUNTU_MIRROR} + + + + ubuntu-security-mirror-set + + + env.UBUNTU_SECURITY_MIRROR + + + + + ${env.UBUNTU_SECURITY_MIRROR} + + + diff --git a/tests/docker-images/java-test-image/pom.xml b/tests/docker-images/java-test-image/pom.xml index ec3cc1a6b4f49..26f13dd3783b3 100644 --- a/tests/docker-images/java-test-image/pom.xml +++ b/tests/docker-images/java-test-image/pom.xml @@ -35,8 +35,8 @@ docker target/pulsar-server-distribution-bin.tar.gz - ${env.UBUNTU_MIRROR} - ${env.UBUNTU_SECURITY_MIRROR} + ${UBUNTU_MIRROR} + ${UBUNTU_SECURITY_MIRROR} From 66333edc2303e86dde4dc860c4bf006422c94f3e Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Fri, 5 Apr 2024 09:14:22 -0700 Subject: [PATCH 6/7] [improve][misc] Specify valid home dir for the default user in the Ubuntu based docker image (#22446) (cherry picked from commit e008580f0fd00451fb9b05caf53757ba6539ba61) (cherry picked from commit c9b656ca8c4485c2e4a9a0d9af7f47a1f909c0cc) --- docker/pulsar/Dockerfile | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docker/pulsar/Dockerfile b/docker/pulsar/Dockerfile index 10bd2b5ee2675..b899fb6182829 100644 --- a/docker/pulsar/Dockerfile +++ b/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 From f8b6bc1ed91cf7bf6d2278936ad111dbde582333 Mon Sep 17 00:00:00 2001 From: nikhil-ctds Date: Mon, 8 Apr 2024 15:39:26 +0530 Subject: [PATCH 7/7] Disabled AlluxioSinkTest --- pom.xml | 3 +++ 1 file changed, 3 insertions(+) diff --git a/pom.xml b/pom.xml index ccb127f2d1066..5c5ddf881896f 100644 --- a/pom.xml +++ b/pom.xml @@ -1616,6 +1616,9 @@ flexible messaging model and an intuitive client API. 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 + + **/AlluxioSink*.java +