You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by te...@apache.org on 2022/08/11 01:39:55 UTC
[pulsar] branch branch-2.11 updated (7c0a7f91546 -> a2769124689)
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a change to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 7c0a7f91546 [improve][test] Avoid hacky modify static final field (#17005)
new 4b3b2f74ca6 [fix][client] Release semaphore before discarding messages in batchMessageContainer (#17019)
new c7ff6d41366 [fix][load-balancer] skip mis-configured resource usage(>100%) in load computation (#16937)
new a2769124689 [test]Fix Flaky-test: BrokerServiceTest.testLookupThrottlingForClientByClient (#16540)
The 3 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:
.../impl/LeastResourceUsageWithWeight.java | 2 +-
.../loadbalance/impl/ModularLoadManagerImpl.java | 10 +--
.../broker/loadbalance/impl/ThresholdShedder.java | 76 +++++++++++++++++++---
.../loadbalance/impl/ThresholdShedderTest.java | 6 ++
.../pulsar/broker/service/BrokerServiceTest.java | 71 ++++++++++++++++----
.../apache/pulsar/client/impl/ProducerImpl.java | 2 +-
.../data/loadbalancer/LocalBrokerData.java | 20 ++++++
7 files changed, 160 insertions(+), 27 deletions(-)
[pulsar] 01/03: [fix][client] Release semaphore before discarding messages in batchMessageContainer (#17019)
Posted by te...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 4b3b2f74ca68a2f08714978b4027ddcf6c0806b4
Author: Zike Yang <zi...@apache.org>
AuthorDate: Thu Aug 11 09:24:30 2022 +0800
[fix][client] Release semaphore before discarding messages in batchMessageContainer (#17019)
---
.../src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
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 b5756e61b2c..210fc14eb66 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
@@ -2030,9 +2030,9 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
}
final int numMessagesInBatch = batchMessageContainer.getNumMessagesInBatch();
final long currentBatchSize = batchMessageContainer.getCurrentBatchSize();
- batchMessageContainer.discard(ex);
semaphoreRelease(numMessagesInBatch);
client.getMemoryLimitController().releaseMemory(currentBatchSize);
+ batchMessageContainer.discard(ex);
}
@Override
[pulsar] 02/03: [fix][load-balancer] skip mis-configured resource usage(>100%) in load computation (#16937)
Posted by te...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit c7ff6d4136692a7b7e4ef58d2bba9d630fe90545
Author: Heesung Sohn <10...@users.noreply.github.com>
AuthorDate: Wed Aug 10 17:25:35 2022 -0700
[fix][load-balancer] skip mis-configured resource usage(>100%) in load computation (#16937)
---
.../impl/LeastResourceUsageWithWeight.java | 2 +-
.../loadbalance/impl/ModularLoadManagerImpl.java | 10 +--
.../broker/loadbalance/impl/ThresholdShedder.java | 76 +++++++++++++++++++---
.../loadbalance/impl/ThresholdShedderTest.java | 6 ++
.../data/loadbalancer/LocalBrokerData.java | 20 ++++++
5 files changed, 100 insertions(+), 14 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/LeastResourceUsageWithWeight.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/LeastResourceUsageWithWeight.java
index 2e717500e64..9493fefbad2 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/LeastResourceUsageWithWeight.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/LeastResourceUsageWithWeight.java
@@ -90,7 +90,7 @@ public class LeastResourceUsageWithWeight implements ModularLoadManagerStrategy
ServiceConfiguration conf) {
final double historyPercentage = conf.getLoadBalancerHistoryResourcePercentage();
Double historyUsage = brokerAvgResourceUsageWithWeight.get(broker);
- double resourceUsage = brokerData.getLocalData().getMaxResourceUsageWithWeight(
+ double resourceUsage = brokerData.getLocalData().getMaxResourceUsageWithWeightWithinLimit(
conf.getLoadBalancerCPUResourceWeight(),
conf.getLoadBalancerMemoryResourceWeight(),
conf.getLoadBalancerDirectMemoryResourceWeight(),
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 9444737fb79..0c86602695a 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
@@ -448,8 +448,9 @@ public class ModularLoadManagerImpl implements ModularLoadManager {
long timeSinceLastReportWrittenToStore = System.currentTimeMillis() - localData.getLastUpdate();
if (timeSinceLastReportWrittenToStore > updateMaxIntervalMillis) {
log.info("Writing local data to metadata store because time since last"
- + " update exceeded threshold of {} minutes",
- conf.getLoadBalancerReportUpdateMaxIntervalMinutes());
+ + " update exceeded threshold of {} minutes. ResourceUsage:[{}]",
+ conf.getLoadBalancerReportUpdateMaxIntervalMinutes(),
+ localData.printResourceUsage());
// Always update after surpassing the maximum interval.
return true;
}
@@ -463,9 +464,10 @@ public class ModularLoadManagerImpl implements ModularLoadManager {
percentChange(lastData.getNumBundles(), localData.getNumBundles()))));
if (maxChange > conf.getLoadBalancerReportUpdateThresholdPercentage()) {
log.info("Writing local data to metadata store because maximum change {}% exceeded threshold {}%; "
- + "time since last report written is {} seconds", maxChange,
+ + "time since last report written is {} seconds. ResourceUsage:[{}]", maxChange,
conf.getLoadBalancerReportUpdateThresholdPercentage(),
- timeSinceLastReportWrittenToStore / 1000.0);
+ timeSinceLastReportWrittenToStore / 1000.0,
+ localData.printResourceUsage());
return true;
}
return false;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedder.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedder.java
index 586a2fe1012..05c69062663 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedder.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedder.java
@@ -53,16 +53,38 @@ public class ThresholdShedder implements LoadSheddingStrategy {
private final Multimap<String, String> selectedBundlesCache = ArrayListMultimap.create();
private static final double ADDITIONAL_THRESHOLD_PERCENT_MARGIN = 0.05;
private static final double MB = 1024 * 1024;
+
+ private static final long LOAD_LOG_SAMPLE_DELAY_IN_SEC = 5 * 60; // 5 mins
private final Map<String, Double> brokerAvgResourceUsage = new HashMap<>();
+ private long lastSampledLoadLogTS = 0;
+
+
+ private static int toPercentage(double usage) {
+ return (int) (usage * 100);
+ }
+
+ private boolean canSampleLog() {
+ long now = System.currentTimeMillis() / 1000;
+ boolean sampleLog = now - lastSampledLoadLogTS >= LOAD_LOG_SAMPLE_DELAY_IN_SEC;
+ if (sampleLog) {
+ lastSampledLoadLogTS = now;
+ }
+ return sampleLog;
+ }
@Override
public Multimap<String, String> findBundlesForUnloading(final LoadData loadData, final ServiceConfiguration conf) {
selectedBundlesCache.clear();
+ boolean sampleLog = canSampleLog();
final double threshold = conf.getLoadBalancerBrokerThresholdShedderPercentage() / 100.0;
final Map<String, Long> recentlyUnloadedBundles = loadData.getRecentlyUnloadedBundles();
final double minThroughputThreshold = conf.getLoadBalancerBundleUnloadMinThroughputThreshold() * MB;
- final double avgUsage = getBrokerAvgUsage(loadData, conf.getLoadBalancerHistoryResourcePercentage(), conf);
+ final double avgUsage = getBrokerAvgUsage(
+ loadData, conf.getLoadBalancerHistoryResourcePercentage(), conf, sampleLog);
+ if (sampleLog) {
+ log.info("brokers' resource avgUsage:{}%", toPercentage(avgUsage));
+ }
if (avgUsage == 0) {
log.warn("average max resource usage is 0");
@@ -74,8 +96,9 @@ public class ThresholdShedder implements LoadSheddingStrategy {
final double currentUsage = brokerAvgResourceUsage.getOrDefault(broker, 0.0);
if (currentUsage < avgUsage + threshold) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] broker is not overloaded, ignoring at this point", broker);
+ if (sampleLog) {
+ log.info("[{}] broker is not overloaded, ignoring at this point, currentUsage:{}%",
+ broker, toPercentage(currentUsage));
}
return;
}
@@ -86,14 +109,13 @@ public class ThresholdShedder implements LoadSheddingStrategy {
double minimumThroughputToOffload = brokerCurrentThroughput * percentOfTrafficToOffload;
if (minimumThroughputToOffload < minThroughputThreshold) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] broker is planning to shed throughput {} MByte/s less than "
+ if (sampleLog) {
+ log.info("[{}] broker is planning to shed throughput {} MByte/s less than "
+ "minimumThroughputThreshold {} MByte/s, skipping bundle unload.",
broker, minimumThroughputToOffload / MB, minThroughputThreshold / MB);
}
return;
}
-
log.info(
"Attempting to shed load on {}, which has max resource usage above avgUsage and threshold {}%"
+ " > {}% + {}% -- Offloading at least {} MByte/s of traffic, left throughput {} MByte/s",
@@ -139,14 +161,14 @@ public class ThresholdShedder implements LoadSheddingStrategy {
}
private double getBrokerAvgUsage(final LoadData loadData, final double historyPercentage,
- final ServiceConfiguration conf) {
+ final ServiceConfiguration conf, boolean sampleLog) {
double totalUsage = 0.0;
int totalBrokers = 0;
for (Map.Entry<String, BrokerData> entry : loadData.getBrokerData().entrySet()) {
LocalBrokerData localBrokerData = entry.getValue().getLocalData();
String broker = entry.getKey();
- totalUsage += updateAvgResourceUsage(broker, localBrokerData, historyPercentage, conf);
+ totalUsage += updateAvgResourceUsage(broker, localBrokerData, historyPercentage, conf, sampleLog);
totalBrokers++;
}
@@ -154,7 +176,8 @@ public class ThresholdShedder implements LoadSheddingStrategy {
}
private double updateAvgResourceUsage(String broker, LocalBrokerData localBrokerData,
- final double historyPercentage, final ServiceConfiguration conf) {
+ final double historyPercentage, final ServiceConfiguration conf,
+ boolean sampleLog) {
Double historyUsage =
brokerAvgResourceUsage.get(broker);
double resourceUsage = localBrokerData.getMaxResourceUsageWithWeight(
@@ -162,6 +185,41 @@ public class ThresholdShedder implements LoadSheddingStrategy {
conf.getLoadBalancerMemoryResourceWeight(), conf.getLoadBalancerDirectMemoryResourceWeight(),
conf.getLoadBalancerBandwithInResourceWeight(),
conf.getLoadBalancerBandwithOutResourceWeight());
+
+ if (sampleLog) {
+ log.info("{} broker load: historyUsage={}%, resourceUsage={}%",
+ broker,
+ historyUsage == null ? 0 : toPercentage(historyUsage),
+ toPercentage(resourceUsage));
+ }
+
+ // wrap if resourceUsage is bigger than 1.0
+ if (resourceUsage > 1.0) {
+ log.error("{} broker resourceUsage is bigger than 100%. "
+ + "Some of the resource limits are mis-configured. "
+ + "Try to disable the error resource signals by setting their weights to zero "
+ + "or fix the resource limit configurations. "
+ + "Ref:https://pulsar.apache.org/docs/administration-load-balance/#thresholdshedder "
+ + "ResourceUsage:[{}], "
+ + "CPUResourceWeight:{}, MemoryResourceWeight:{}, DirectMemoryResourceWeight:{}, "
+ + "BandwithInResourceWeight:{}, BandwithOutResourceWeight:{}",
+ broker,
+ localBrokerData.printResourceUsage(),
+ conf.getLoadBalancerCPUResourceWeight(),
+ conf.getLoadBalancerMemoryResourceWeight(),
+ conf.getLoadBalancerDirectMemoryResourceWeight(),
+ conf.getLoadBalancerBandwithInResourceWeight(),
+ conf.getLoadBalancerBandwithOutResourceWeight());
+
+ resourceUsage = localBrokerData.getMaxResourceUsageWithWeightWithinLimit(
+ conf.getLoadBalancerCPUResourceWeight(),
+ conf.getLoadBalancerMemoryResourceWeight(), conf.getLoadBalancerDirectMemoryResourceWeight(),
+ conf.getLoadBalancerBandwithInResourceWeight(),
+ conf.getLoadBalancerBandwithOutResourceWeight());
+
+ log.warn("{} broker recomputed max resourceUsage={}%. Skipped usage signals bigger than 100%",
+ broker, toPercentage(resourceUsage));
+ }
historyUsage = historyUsage == null
? resourceUsage : historyUsage * historyPercentage + (1 - historyPercentage) * resourceUsage;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedderTest.java
index 4af7d784909..80760a7afc3 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedderTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ThresholdShedderTest.java
@@ -73,6 +73,9 @@ public class ThresholdShedderTest {
LoadData loadData = new LoadData();
LocalBrokerData broker1 = new LocalBrokerData();
+ broker1.setCpu(new ResourceUsage(1000, 100));
+ broker1.setMemory(new ResourceUsage(5000, 100));
+ broker1.setDirectMemory(new ResourceUsage(5000, 100));
broker1.setBandwidthIn(new ResourceUsage(500, 1000));
broker1.setBandwidthOut(new ResourceUsage(500, 1000));
broker1.setBundles(Sets.newHashSet("bundle-1"));
@@ -115,6 +118,9 @@ public class ThresholdShedderTest {
LoadData loadData = new LoadData();
LocalBrokerData broker1 = new LocalBrokerData();
+ broker1.setCpu(new ResourceUsage(1000, 100));
+ broker1.setMemory(new ResourceUsage(5000, 100));
+ broker1.setDirectMemory(new ResourceUsage(5000, 100));
broker1.setBandwidthIn(new ResourceUsage(999, 1000));
broker1.setBandwidthOut(new ResourceUsage(999, 1000));
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/policies/data/loadbalancer/LocalBrokerData.java b/pulsar-common/src/main/java/org/apache/pulsar/policies/data/loadbalancer/LocalBrokerData.java
index e63de9b0dca..8736b675272 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/policies/data/loadbalancer/LocalBrokerData.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/policies/data/loadbalancer/LocalBrokerData.java
@@ -255,6 +255,16 @@ public class LocalBrokerData implements LoadManagerReport {
bandwidthOut.percentUsage() * bandwidthOutWeight) / 100;
}
+ public double getMaxResourceUsageWithWeightWithinLimit(final double cpuWeight, final double memoryWeight,
+ final double directMemoryWeight,
+ final double bandwidthInWeight,
+ final double bandwidthOutWeight) {
+ return maxWithinLimit(100.0d,
+ cpu.percentUsage() * cpuWeight, memory.percentUsage() * memoryWeight,
+ directMemory.percentUsage() * directMemoryWeight, bandwidthIn.percentUsage() * bandwidthInWeight,
+ bandwidthOut.percentUsage() * bandwidthOutWeight) / 100;
+ }
+
private static double max(double... args) {
double max = Double.NEGATIVE_INFINITY;
@@ -279,6 +289,16 @@ public class LocalBrokerData implements LoadManagerReport {
return max;
}
+ private static double maxWithinLimit(double limit, double...args) {
+ double max = 0.0;
+ for (double d : args) {
+ if (d > max && d <= limit) {
+ max = d;
+ }
+ }
+ return max;
+ }
+
public String getLoadReportType() {
return loadReportType;
}
[pulsar] 03/03: [test]Fix Flaky-test: BrokerServiceTest.testLookupThrottlingForClientByClient (#16540)
Posted by te...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit a2769124689892bd5153f9b5b2e5ac20c1121038
Author: Xiaoyu Hou <An...@gmail.com>
AuthorDate: Thu Aug 11 00:06:54 2022 +0800
[test]Fix Flaky-test: BrokerServiceTest.testLookupThrottlingForClientByClient (#16540)
---
.../pulsar/broker/service/BrokerServiceTest.java | 71 ++++++++++++++++++----
1 file changed, 59 insertions(+), 12 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java
index faa5de072aa..1fb63470456 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java
@@ -61,6 +61,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
@@ -88,10 +89,13 @@ import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.client.impl.ClientCnx;
import org.apache.pulsar.client.impl.ConnectionPool;
import org.apache.pulsar.client.impl.PulsarServiceNameResolver;
import org.apache.pulsar.client.impl.auth.AuthenticationTls;
import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.common.api.proto.CommandLookupTopicResponse;
+import org.apache.pulsar.common.api.proto.CommandPartitionedTopicMetadataResponse;
import org.apache.pulsar.common.naming.NamespaceBundle;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicName;
@@ -966,8 +970,6 @@ public class BrokerServiceTest extends BrokerTestBase {
*/
@Test
public void testLookupThrottlingForClientByClient() throws Exception {
- // This test looks like it could be flakey, if the broker responds
- // quickly enough, there may never be concurrency in requests
final String topicName = "persistent://prop/ns-abc/newTopic";
PulsarServiceNameResolver resolver = new PulsarServiceNameResolver();
@@ -979,7 +981,30 @@ public class BrokerServiceTest extends BrokerTestBase {
EventLoopGroup eventLoop = EventLoopUtil.newEventLoopGroup(20, false,
new DefaultThreadFactory("test-pool", Thread.currentThread().isDaemon()));
long reqId = 0xdeadbeef;
- try (ConnectionPool pool = new ConnectionPool(conf, eventLoop)) {
+ // Using an AtomicReference in order to reset a new CountDownLatch
+ AtomicReference<CountDownLatch> latchRef = new AtomicReference<>();
+ latchRef.set(new CountDownLatch(1));
+ try (ConnectionPool pool = new ConnectionPool(conf, eventLoop, () -> new ClientCnx(conf, eventLoop) {
+ @Override
+ protected void handleLookupResponse(CommandLookupTopicResponse lookupResult) {
+ try {
+ latchRef.get().await();
+ } catch (InterruptedException e) {
+ // ignore
+ }
+ super.handleLookupResponse(lookupResult);
+ }
+
+ @Override
+ protected void handlePartitionResponse(CommandPartitionedTopicMetadataResponse lookupResult) {
+ try {
+ latchRef.get().await();
+ } catch (InterruptedException e) {
+ // ignore
+ }
+ super.handlePartitionResponse(lookupResult);
+ }
+ })) {
// for PMR
// 2 lookup will succeed
long reqId1 = reqId++;
@@ -990,12 +1015,18 @@ public class BrokerServiceTest extends BrokerTestBase {
long reqId2 = reqId++;
ByteBuf request2 = Commands.newPartitionMetadataRequest(topicName, reqId2);
CompletableFuture<?> f2 = pool.getConnection(resolver.resolveHost())
- .thenCompose(clientCnx -> clientCnx.newLookup(request2, reqId2));
+ .thenCompose(clientCnx -> {
+ CompletableFuture<?> future = clientCnx.newLookup(request2, reqId2);
+ // pending other responses in `ClientCnx` until now
+ latchRef.get().countDown();
+ return future;
+ });
f1.get();
f2.get();
// 3 lookup will fail
+ latchRef.set(new CountDownLatch(1));
long reqId3 = reqId++;
ByteBuf request3 = Commands.newPartitionMetadataRequest(topicName, reqId3);
f1 = pool.getConnection(resolver.resolveHost())
@@ -1009,11 +1040,16 @@ public class BrokerServiceTest extends BrokerTestBase {
long reqId5 = reqId++;
ByteBuf request5 = Commands.newPartitionMetadataRequest(topicName, reqId5);
CompletableFuture<?> f3 = pool.getConnection(resolver.resolveHost())
- .thenCompose(clientCnx -> clientCnx.newLookup(request5, reqId5));
+ .thenCompose(clientCnx -> {
+ CompletableFuture<?> future = clientCnx.newLookup(request5, reqId5);
+ // pending other responses in `ClientCnx` until now
+ latchRef.get().countDown();
+ return future;
+ });
+ f1.get();
+ f2.get();
try {
- f1.get();
- f2.get();
f3.get();
fail("At least one should fail");
} catch (ExecutionException e) {
@@ -1029,6 +1065,7 @@ public class BrokerServiceTest extends BrokerTestBase {
// for Lookup
// 2 lookup will succeed
+ latchRef.set(new CountDownLatch(1));
long reqId6 = reqId++;
ByteBuf request6 = Commands.newLookup(topicName, true, reqId6);
f1 = pool.getConnection(resolver.resolveHost())
@@ -1037,12 +1074,18 @@ public class BrokerServiceTest extends BrokerTestBase {
long reqId7 = reqId++;
ByteBuf request7 = Commands.newLookup(topicName, true, reqId7);
f2 = pool.getConnection(resolver.resolveHost())
- .thenCompose(clientCnx -> clientCnx.newLookup(request7, reqId7));
+ .thenCompose(clientCnx -> {
+ CompletableFuture<?> future = clientCnx.newLookup(request7, reqId7);
+ // pending other responses in `ClientCnx` until now
+ latchRef.get().countDown();
+ return future;
+ });
f1.get();
f2.get();
// 3 lookup will fail
+ latchRef.set(new CountDownLatch(1));
long reqId8 = reqId++;
ByteBuf request8 = Commands.newLookup(topicName, true, reqId8);
f1 = pool.getConnection(resolver.resolveHost())
@@ -1056,11 +1099,16 @@ public class BrokerServiceTest extends BrokerTestBase {
long reqId10 = reqId++;
ByteBuf request10 = Commands.newLookup(topicName, true, reqId10);
f3 = pool.getConnection(resolver.resolveHost())
- .thenCompose(clientCnx -> clientCnx.newLookup(request10, reqId10));
+ .thenCompose(clientCnx -> {
+ CompletableFuture<?> future = clientCnx.newLookup(request10, reqId10);
+ // pending other responses in `ClientCnx` until now
+ latchRef.get().countDown();
+ return future;
+ });
+ f1.get();
+ f2.get();
try {
- f1.get();
- f2.get();
f3.get();
fail("At least one should fail");
} catch (ExecutionException e) {
@@ -1073,7 +1121,6 @@ public class BrokerServiceTest extends BrokerTestBase {
throw e;
}
}
-
}
}