You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "merlimat (via GitHub)" <gi...@apache.org> on 2026/09/17 01:33:57 UTC
[PR] [fix][client] Keep a gone segment's producer until the layout drops it [pulsar]
merlimat opened a new pull request, #26617:
URL: https://github.com/apache/pulsar/pull/26617
Stacked on #26615 (its commit is included here; the diff shrinks to the second commit once it merges).
### Motivation
A perf-producer run at ~1M msg/s on a scalable topic produced a 10 MB log for a single segment split. Two storms, one root cause:
1. **At the split.** The broker terminates the parent segment's topic before it publishes the new layout. Every send that finds the segment gone is retried once the DAG watch delivers the new layout, and each such retry dropped the segment's v4 producer from `ScalableTopicProducer`'s map. Every send routed to the segment in the meantime then created a fresh v4 producer on the terminated topic: 2,958 producer creations within one second, each logging `Creating producer`, the broker's `TopicTerminated` rejection and `Temporary error in creating producer`.
2. **At Ctrl-C.** Closing the client closes the v4 segment producers, which fail their pending sends with `AlreadyClosed`. That is classified as "segment gone", so the same path ran: 1,112 producers created on a closing client, each reconnecting on backoff (`Reconnecting after connection was closed` / `Error connecting to broker`, ~11,000 lines), and the 243 sends in flight only failed once the 10-attempt retry budget ran out, four seconds after the close.
### Modifications
- `ScalableTopicProducer` leaves a dead segment producer where it is until `applyLayout` drops it, on both the sync and the async retry path. A send routed to the segment before the new layout arrives fails fast on the existing producer and is retried as before; no producer is re-created on a terminated topic. Migration keeps the legacy partitions as sealed parents with fresh ids for the children, so the next layout always removes the dead producer.
- Once the producer or its v4 client is closing, sends are failed with `AlreadyClosedException` instead of retried, on the first attempt and on a retry alike: no new layout is coming.
- A send that raced a layout change, whose segment is no longer in the layout by the time its producer is created (`Segment N not found in active segments`), is now retried like any other gone segment instead of failing the caller's future.
- `applyLayout` no longer warns `Error closing producer for segment` for a producer whose creation had failed: with the producer left in the map, that case is reachable and there is nothing to close.
- A `@VisibleForTesting` counter of segment producer creations, and `V5ProducerSegmentGoneTest`:
- sends keep flowing while a segment's topic is terminated underneath an unchanged layout, the state a split leaves the producer in until the new layout arrives: **1** producer created with this change, **13,754** without it;
- closing the client while sends are in flight fails them promptly and creates no producer.
Verified with `V5ProducerBackpressureTest` (including the split accounting test), `V5MigrationEndToEndTest` and the `pulsar-client-v5` unit tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Keep a gone segment's producer until the layout drops it [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #26617:
URL: https://github.com/apache/pulsar/pull/26617#discussion_r4040207804
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/v5/V5ProducerSegmentGoneTest.java:
##########
@@ -0,0 +1,187 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl.v5;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.LockSupport;
+import org.apache.pulsar.client.api.v5.MessageId;
+import org.apache.pulsar.client.api.v5.Producer;
+import org.apache.pulsar.client.api.v5.PulsarClient;
+import org.apache.pulsar.client.api.v5.V5ClientBaseTest;
+import org.apache.pulsar.client.api.v5.config.BatchingPolicy;
+import org.apache.pulsar.client.api.v5.config.MemorySize;
+import org.apache.pulsar.client.api.v5.schema.Schema;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.scalable.HashRange;
+import org.apache.pulsar.common.scalable.SegmentTopicName;
+import org.awaitility.Awaitility;
+import org.testng.annotations.Test;
+
+/**
+ * A send that finds its target segment gone waits for the next layout and retries. While it waits,
+ * the per-segment producer that is gone must stay where it is: dropping it would have every send
+ * routed to the segment in the meantime create a new v4 producer on a terminated topic, thousands
+ * of them a second under load. Once the producer or its client is closing, no layout is coming and
+ * the sends fail instead of waiting for one.
+ */
+public class V5ProducerSegmentGoneTest extends V5ClientBaseTest {
+
+ /** Sends per burst; a burst is followed by a millisecond pause, a pace the broker keeps up with. */
+ private static final int BURST = 20;
+
+ /** The sends must keep flowing while a segment is sealed, so the limit must not block the sender. */
+ private PulsarClient newClient() throws Exception {
+ return track(PulsarClient.builder()
+ .serviceUrl(getBrokerServiceUrl())
+ .memoryLimit(MemorySize.ofBytes(256 * 1024 * 1024))
+ .build());
+ }
+
+ private ScalableTopicProducer<byte[]> newProducer(PulsarClient client, String topic) throws Exception {
+ return (ScalableTopicProducer<byte[]>) track(client.newProducer(Schema.bytes())
+ .topic(topic)
+ .batchingPolicy(BatchingPolicy.ofDisabled())
+ .create());
+ }
+
+ /** The sends a {@link #startSender sender} thread has issued and what became of them. */
+ private static final class Sends {
+ final List<CompletableFuture<MessageId>> futures = new ArrayList<>();
+ final AtomicInteger issued = new AtomicInteger();
+ final AtomicInteger completed = new AtomicInteger();
+ final AtomicInteger failed = new AtomicInteger();
+ final Set<Long> segmentsAcked = ConcurrentHashMap.newKeySet();
+ final AtomicBoolean pause = new AtomicBoolean();
+ final AtomicBoolean stop = new AtomicBoolean();
+
+ boolean drained() {
+ return completed.get() == issued.get();
+ }
+ }
+
+ /** Send paced bursts, pausing on request, until told to stop. */
+ private static Thread startSender(Producer<byte[]> producer, Sends sends) {
+ byte[] payload = new byte[1024];
+ Thread sender = new Thread(() -> {
+ while (!sends.stop.get()) {
+ if (!sends.pause.get()) {
+ for (int i = 0; i < BURST; i++) {
+ CompletableFuture<MessageId> future = producer.async().newMessage().value(payload).send();
+ future.whenComplete((id, ex) -> {
+ if (ex == null) {
+ sends.segmentsAcked.add(((MessageIdV5) id).segmentId());
+ } else {
+ sends.failed.incrementAndGet();
+ }
+ sends.completed.incrementAndGet();
+ });
+ sends.futures.add(future);
+ sends.issued.incrementAndGet();
+ }
+ }
+ LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(1));
+ }
+ }, "sender");
+ sender.start();
+ return sender;
+ }
+
+ private static CompletableFuture<Void> allSettled(Sends sends) {
+ return CompletableFuture.allOf(sends.futures.toArray(new CompletableFuture[0]))
+ .exceptionally(__ -> null);
+ }
+
+ private long singleActiveSegmentId(String topic) throws Exception {
+ for (var segment : admin.scalableTopics().getMetadata(topic).getSegments().values()) {
+ if (segment.isActive()) {
+ return segment.getSegmentId();
+ }
+ }
+ throw new AssertionError("no active segment for " + topic);
+ }
+
+ private String segmentTopic(String topic, long segmentId) throws Exception {
+ var segment = admin.scalableTopics().getMetadata(topic).getSegments().get(segmentId);
+ HashRange range = HashRange.of(segment.getHashRange().getStart(), segment.getHashRange().getEnd());
+ return SegmentTopicName.fromParent(TopicName.get(topic), range, segmentId).toString();
+ }
+
+ @Test
+ public void aSealedSegmentGetsNoNewProducerForTheSendsThatFindItGone() throws Exception {
+ PulsarClient client = newClient();
+ String topic = newScalableTopic(1);
+ long segmentId = singleActiveSegmentId(topic);
+ ScalableTopicProducer<byte[]> producer = newProducer(client, topic);
+
+ Sends sends = new Sends();
+ Thread sender = startSender(producer, sends);
Review Comment:
Thanks, this addresses the cleanup request in both tests. The finally blocks cover the failing waits and broker/client operations, and [`V5ProducerSegmentGoneTest.java:115-123`](https://github.com/apache/pulsar/blob/58dbd531d5aeb677171012e8242877b8765b6933/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/v5/V5ProducerSegmentGoneTest.java#L115-L123) stops the sender with bounded joins and interruption if needed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Keep a gone segment's producer until the layout drops it [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #26617:
URL: https://github.com/apache/pulsar/pull/26617#discussion_r4033563353
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/v5/V5ProducerSegmentGoneTest.java:
##########
@@ -0,0 +1,187 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl.v5;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.LockSupport;
+import org.apache.pulsar.client.api.v5.MessageId;
+import org.apache.pulsar.client.api.v5.Producer;
+import org.apache.pulsar.client.api.v5.PulsarClient;
+import org.apache.pulsar.client.api.v5.V5ClientBaseTest;
+import org.apache.pulsar.client.api.v5.config.BatchingPolicy;
+import org.apache.pulsar.client.api.v5.config.MemorySize;
+import org.apache.pulsar.client.api.v5.schema.Schema;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.scalable.HashRange;
+import org.apache.pulsar.common.scalable.SegmentTopicName;
+import org.awaitility.Awaitility;
+import org.testng.annotations.Test;
+
+/**
+ * A send that finds its target segment gone waits for the next layout and retries. While it waits,
+ * the per-segment producer that is gone must stay where it is: dropping it would have every send
+ * routed to the segment in the meantime create a new v4 producer on a terminated topic, thousands
+ * of them a second under load. Once the producer or its client is closing, no layout is coming and
+ * the sends fail instead of waiting for one.
+ */
+public class V5ProducerSegmentGoneTest extends V5ClientBaseTest {
+
+ /** Sends per burst; a burst is followed by a millisecond pause, a pace the broker keeps up with. */
+ private static final int BURST = 20;
+
+ /** The sends must keep flowing while a segment is sealed, so the limit must not block the sender. */
+ private PulsarClient newClient() throws Exception {
+ return track(PulsarClient.builder()
+ .serviceUrl(getBrokerServiceUrl())
+ .memoryLimit(MemorySize.ofBytes(256 * 1024 * 1024))
+ .build());
+ }
+
+ private ScalableTopicProducer<byte[]> newProducer(PulsarClient client, String topic) throws Exception {
+ return (ScalableTopicProducer<byte[]>) track(client.newProducer(Schema.bytes())
+ .topic(topic)
+ .batchingPolicy(BatchingPolicy.ofDisabled())
+ .create());
+ }
+
+ /** The sends a {@link #startSender sender} thread has issued and what became of them. */
+ private static final class Sends {
+ final List<CompletableFuture<MessageId>> futures = new ArrayList<>();
+ final AtomicInteger issued = new AtomicInteger();
+ final AtomicInteger completed = new AtomicInteger();
+ final AtomicInteger failed = new AtomicInteger();
+ final Set<Long> segmentsAcked = ConcurrentHashMap.newKeySet();
+ final AtomicBoolean pause = new AtomicBoolean();
+ final AtomicBoolean stop = new AtomicBoolean();
+
+ boolean drained() {
+ return completed.get() == issued.get();
+ }
+ }
+
+ /** Send paced bursts, pausing on request, until told to stop. */
+ private static Thread startSender(Producer<byte[]> producer, Sends sends) {
+ byte[] payload = new byte[1024];
+ Thread sender = new Thread(() -> {
+ while (!sends.stop.get()) {
+ if (!sends.pause.get()) {
+ for (int i = 0; i < BURST; i++) {
+ CompletableFuture<MessageId> future = producer.async().newMessage().value(payload).send();
+ future.whenComplete((id, ex) -> {
+ if (ex == null) {
+ sends.segmentsAcked.add(((MessageIdV5) id).segmentId());
+ } else {
+ sends.failed.incrementAndGet();
+ }
+ sends.completed.incrementAndGet();
+ });
+ sends.futures.add(future);
+ sends.issued.incrementAndGet();
+ }
+ }
+ LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(1));
+ }
+ }, "sender");
+ sender.start();
+ return sender;
+ }
+
+ private static CompletableFuture<Void> allSettled(Sends sends) {
+ return CompletableFuture.allOf(sends.futures.toArray(new CompletableFuture[0]))
+ .exceptionally(__ -> null);
+ }
+
+ private long singleActiveSegmentId(String topic) throws Exception {
+ for (var segment : admin.scalableTopics().getMetadata(topic).getSegments().values()) {
+ if (segment.isActive()) {
+ return segment.getSegmentId();
+ }
+ }
+ throw new AssertionError("no active segment for " + topic);
+ }
+
+ private String segmentTopic(String topic, long segmentId) throws Exception {
+ var segment = admin.scalableTopics().getMetadata(topic).getSegments().get(segmentId);
+ HashRange range = HashRange.of(segment.getHashRange().getStart(), segment.getHashRange().getEnd());
+ return SegmentTopicName.fromParent(TopicName.get(topic), range, segmentId).toString();
+ }
+
+ @Test
+ public void aSealedSegmentGetsNoNewProducerForTheSendsThatFindItGone() throws Exception {
+ PulsarClient client = newClient();
+ String topic = newScalableTopic(1);
+ long segmentId = singleActiveSegmentId(topic);
+ ScalableTopicProducer<byte[]> producer = newProducer(client, topic);
+
+ Sends sends = new Sends();
+ Thread sender = startSender(producer, sends);
Review Comment:
**[QUALITY] Stop the sender in a finally block in both tests**
If an await times out or terminateSegment throws, execution skips the stop/join below. The same applies to the second test. The fixture closes the producer and client, but the sender keeps looping and retaining failed send futures in sends.futures, so one test failure can leak a thread and steadily consume the test JVM heap. Please wrap each sender lifetime in try/finally that sets stop, interrupts it if needed, and joins with a timeout.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Keep a gone segment's producer until the layout drops it [pulsar]
Posted by "merlimat (via GitHub)" <gi...@apache.org>.
merlimat merged PR #26617:
URL: https://github.com/apache/pulsar/pull/26617
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Keep a gone segment's producer until the layout drops it [pulsar]
Posted by "merlimat (via GitHub)" <gi...@apache.org>.
merlimat commented on code in PR #26617:
URL: https://github.com/apache/pulsar/pull/26617#discussion_r4039300402
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/v5/V5ProducerSegmentGoneTest.java:
##########
@@ -0,0 +1,187 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl.v5;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.LockSupport;
+import org.apache.pulsar.client.api.v5.MessageId;
+import org.apache.pulsar.client.api.v5.Producer;
+import org.apache.pulsar.client.api.v5.PulsarClient;
+import org.apache.pulsar.client.api.v5.V5ClientBaseTest;
+import org.apache.pulsar.client.api.v5.config.BatchingPolicy;
+import org.apache.pulsar.client.api.v5.config.MemorySize;
+import org.apache.pulsar.client.api.v5.schema.Schema;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.scalable.HashRange;
+import org.apache.pulsar.common.scalable.SegmentTopicName;
+import org.awaitility.Awaitility;
+import org.testng.annotations.Test;
+
+/**
+ * A send that finds its target segment gone waits for the next layout and retries. While it waits,
+ * the per-segment producer that is gone must stay where it is: dropping it would have every send
+ * routed to the segment in the meantime create a new v4 producer on a terminated topic, thousands
+ * of them a second under load. Once the producer or its client is closing, no layout is coming and
+ * the sends fail instead of waiting for one.
+ */
+public class V5ProducerSegmentGoneTest extends V5ClientBaseTest {
+
+ /** Sends per burst; a burst is followed by a millisecond pause, a pace the broker keeps up with. */
+ private static final int BURST = 20;
+
+ /** The sends must keep flowing while a segment is sealed, so the limit must not block the sender. */
+ private PulsarClient newClient() throws Exception {
+ return track(PulsarClient.builder()
+ .serviceUrl(getBrokerServiceUrl())
+ .memoryLimit(MemorySize.ofBytes(256 * 1024 * 1024))
+ .build());
+ }
+
+ private ScalableTopicProducer<byte[]> newProducer(PulsarClient client, String topic) throws Exception {
+ return (ScalableTopicProducer<byte[]>) track(client.newProducer(Schema.bytes())
+ .topic(topic)
+ .batchingPolicy(BatchingPolicy.ofDisabled())
+ .create());
+ }
+
+ /** The sends a {@link #startSender sender} thread has issued and what became of them. */
+ private static final class Sends {
+ final List<CompletableFuture<MessageId>> futures = new ArrayList<>();
+ final AtomicInteger issued = new AtomicInteger();
+ final AtomicInteger completed = new AtomicInteger();
+ final AtomicInteger failed = new AtomicInteger();
+ final Set<Long> segmentsAcked = ConcurrentHashMap.newKeySet();
+ final AtomicBoolean pause = new AtomicBoolean();
+ final AtomicBoolean stop = new AtomicBoolean();
+
+ boolean drained() {
+ return completed.get() == issued.get();
+ }
+ }
+
+ /** Send paced bursts, pausing on request, until told to stop. */
+ private static Thread startSender(Producer<byte[]> producer, Sends sends) {
+ byte[] payload = new byte[1024];
+ Thread sender = new Thread(() -> {
+ while (!sends.stop.get()) {
+ if (!sends.pause.get()) {
+ for (int i = 0; i < BURST; i++) {
+ CompletableFuture<MessageId> future = producer.async().newMessage().value(payload).send();
+ future.whenComplete((id, ex) -> {
+ if (ex == null) {
+ sends.segmentsAcked.add(((MessageIdV5) id).segmentId());
+ } else {
+ sends.failed.incrementAndGet();
+ }
+ sends.completed.incrementAndGet();
+ });
+ sends.futures.add(future);
+ sends.issued.incrementAndGet();
+ }
+ }
+ LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(1));
+ }
+ }, "sender");
+ sender.start();
+ return sender;
+ }
+
+ private static CompletableFuture<Void> allSettled(Sends sends) {
+ return CompletableFuture.allOf(sends.futures.toArray(new CompletableFuture[0]))
+ .exceptionally(__ -> null);
+ }
+
+ private long singleActiveSegmentId(String topic) throws Exception {
+ for (var segment : admin.scalableTopics().getMetadata(topic).getSegments().values()) {
+ if (segment.isActive()) {
+ return segment.getSegmentId();
+ }
+ }
+ throw new AssertionError("no active segment for " + topic);
+ }
+
+ private String segmentTopic(String topic, long segmentId) throws Exception {
+ var segment = admin.scalableTopics().getMetadata(topic).getSegments().get(segmentId);
+ HashRange range = HashRange.of(segment.getHashRange().getStart(), segment.getHashRange().getEnd());
+ return SegmentTopicName.fromParent(TopicName.get(topic), range, segmentId).toString();
+ }
+
+ @Test
+ public void aSealedSegmentGetsNoNewProducerForTheSendsThatFindItGone() throws Exception {
+ PulsarClient client = newClient();
+ String topic = newScalableTopic(1);
+ long segmentId = singleActiveSegmentId(topic);
+ ScalableTopicProducer<byte[]> producer = newProducer(client, topic);
+
+ Sends sends = new Sends();
+ Thread sender = startSender(producer, sends);
Review Comment:
Addressed in 58dbd531d5a: both tests now run the sender inside try/finally, and `stopSender` sets the stop flag, joins with a 10 s timeout, and interrupts the thread only if it is still alive (blocked in a send). Also merged master to resolve the conflict with #26615, which had merged as a squash.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org