You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/06/01 17:35:59 UTC
(pulsar) branch branch-4.2 updated (6334d191ed1 -> b3c7f394c86)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 6334d191ed1 [improve][misc] Upgrade Netty to 4.1.134 (#25870)
new d93426e0a45 [fix][test] Stabilize WebService rate limiting test (#25866)
new 5991a1d685e [fix][fn] Fix Go function runtime to continue after user exceptions and add neg-ack tests (#25867)
new 5f10bf0c96f [fix][test] Fix flaky ProducerCleanupTest timer cleanup (#25864)
new b3c7f394c86 [fix][test] Fix flaky PulsarFunctionTlsTest.testFunctionsCreation() test (#25889)
The 4 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:
.../apache/pulsar/broker/web/WebServiceTest.java | 24 +++++++++++--
.../pulsar/client/api/ProducerCleanupTest.java | 17 +++++----
.../functions/worker/PulsarFunctionTlsTest.java | 37 ++++++++++---------
pulsar-function-go/pf/instance.go | 41 +++++++++++++++-------
pulsar-function-go/pf/instance_test.go | 32 +++++++++++++++++
.../docker-images/latest-version-image/Dockerfile | 1 +
.../go-examples/exceptionFunc/exceptionFunc.go | 11 ++++--
.../integration/functions/PulsarFunctionsTest.java | 4 +++
.../functions/PulsarFunctionsTestBase.java | 1 +
.../functions/go/PulsarFunctionsGoTest.java | 5 +++
10 files changed, 131 insertions(+), 42 deletions(-)
copy pulsar-function-go/examples/exclamationFunc/exclamationFunc.go => tests/docker-images/latest-version-image/go-examples/exceptionFunc/exceptionFunc.go (84%)
(pulsar) 01/04: [fix][test] Stabilize WebService rate limiting test (#25866)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit d93426e0a45091e77d3bb52d4a8ac1f1fc175352
Author: Oneby Wang <44...@users.noreply.github.com>
AuthorDate: Mon May 25 15:06:40 2026 +0800
[fix][test] Stabilize WebService rate limiting test (#25866)
(cherry picked from commit 3b6a76096740870e87c93919b21ddbd1ee2cdeed)
---
.../apache/pulsar/broker/web/WebServiceTest.java | 24 ++++++++++++++++++++--
1 file changed, 22 insertions(+), 2 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/WebServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/WebServiceTest.java
index 0f9ccdf4af7..c307eb9b1ae 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/WebServiceTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/web/WebServiceTest.java
@@ -269,7 +269,8 @@ public class WebServiceTest {
@Test
public void testRateLimiting() throws Exception {
- setupEnv(false, false, false, false, 10.0, false);
+ double rateLimit = 10.0;
+ setupEnv(false, false, false, false, rateLimit, false);
// setupEnv makes HTTP calls to create the cluster, tenant, and namespace.
var metrics = pulsarTestContext.getOpenTelemetryMetricReader().collectAllMetrics();
@@ -283,7 +284,7 @@ public class WebServiceTest {
// Make requests without exceeding the max rate
for (int i = 0; i < 5; i++) {
makeHttpRequest(false, false);
- Thread.sleep(200);
+ Thread.sleep(rateLimitPauseMillis(rateLimit));
}
metrics = pulsarTestContext.getOpenTelemetryMetricReader().collectAllMetrics();
@@ -573,13 +574,32 @@ public class WebServiceTest {
} catch (ConflictException ce) {
// This is OK.
}
+ sleepForRateLimiter(rateLimit);
+
try {
pulsarAdmin.tenants().createTenant("my-property",
TenantInfo.builder().allowedClusters(Sets.newHashSet(config.getClusterName())).build());
+ } catch (Exception e) {
+ // This is OK.
+ }
+ sleepForRateLimiter(rateLimit);
+
+ try {
pulsarAdmin.namespaces().createNamespace("my-property/my-namespace");
} catch (Exception e) {
// This is OK.
}
+ sleepForRateLimiter(rateLimit);
+ }
+
+ private static void sleepForRateLimiter(double rateLimit) throws InterruptedException {
+ if (rateLimit > 0) {
+ Thread.sleep(rateLimitPauseMillis(rateLimit));
+ }
+ }
+
+ private static long rateLimitPauseMillis(double rateLimit) {
+ return (long) Math.ceil((1000.0 / rateLimit) * 2);
}
@AfterMethod(alwaysRun = true)
(pulsar) 04/04: [fix][test] Fix flaky PulsarFunctionTlsTest.testFunctionsCreation() test (#25889)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit b3c7f394c86b43bcd86fc2172a58b90e622b394a
Author: Oneby Wang <44...@users.noreply.github.com>
AuthorDate: Sat May 30 16:16:16 2026 +0800
[fix][test] Fix flaky PulsarFunctionTlsTest.testFunctionsCreation() test (#25889)
(cherry picked from commit a6af80198f26d31e991e880db3f9b7d601a579c2)
---
.../functions/worker/PulsarFunctionTlsTest.java | 37 +++++++++++-----------
1 file changed, 18 insertions(+), 19 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionTlsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionTlsTest.java
index e23dd51dc4a..7872b795667 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionTlsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionTlsTest.java
@@ -20,8 +20,6 @@ package org.apache.pulsar.functions.worker;
import static org.apache.pulsar.common.util.PortManager.nextLockedFreePort;
import static org.testng.Assert.assertEquals;
-import static org.testng.Assert.assertNotNull;
-import static org.testng.Assert.assertTrue;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.Sets;
@@ -42,10 +40,10 @@ import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.authentication.AuthenticationProviderTls;
import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.api.Authentication;
import org.apache.pulsar.client.impl.auth.AuthenticationTls;
import org.apache.pulsar.common.functions.FunctionConfig;
-import org.apache.pulsar.common.functions.WorkerInfo;
import org.apache.pulsar.common.policies.data.TenantInfo;
import org.apache.pulsar.common.util.ClassLoaderUtils;
import org.apache.pulsar.common.util.ObjectMapperFactory;
@@ -260,23 +258,16 @@ public class PulsarFunctionTlsTest {
log.info(" -------- Start test function : {}", functionName);
- int finalI = i;
- // Wait for a leader to be ready and create the function.
- // The createFunctionWithUrl call is included in the retry loop because a leadership
- // transition can happen between the leader check and the actual API call, causing
- // a 503 "Leader not yet ready" error.
final PulsarAdmin createAdmin = pulsarAdmins[i];
- Awaitility.await().atMost(1, TimeUnit.MINUTES).pollInterval(1, TimeUnit.SECONDS).untilAsserted(() -> {
- final PulsarWorkerService workerService = ((PulsarWorkerService) fnWorkerServices[finalI]);
- final LeaderService leaderService = workerService.getLeaderService();
- assertNotNull(leaderService);
- if (!leaderService.isLeader()) {
- final WorkerInfo workerInfo = workerService.getMembershipManager().getLeader();
- assertTrue(workerInfo != null
- && !workerInfo.getWorkerId().equals(workerService.getWorkerConfig().getWorkerId()));
- }
- createAdmin.functions().createFunctionWithUrl(functionConfig, jarFilePathUrl);
- });
+ // During function-worker leadership election/switchover, the coordination topic can already point to
+ // the new leader while that worker is still finishing its leader initialization. In that short window
+ // the internal /functions/leader request returns a transient 503 "Leader not yet ready", so retry only
+ // that condition and let all other failures surface immediately.
+ Awaitility.await().atMost(1, TimeUnit.MINUTES)
+ .pollInterval(1, TimeUnit.SECONDS)
+ .ignoreExceptionsMatching(PulsarFunctionTlsTest::isLeaderNotReady)
+ .untilAsserted(() -> createAdmin.functions()
+ .createFunctionWithUrl(functionConfig, jarFilePathUrl));
// Function creation is not strongly consistent, so this test can fail with a get that is too eager and
// does not have retries.
@@ -293,6 +284,14 @@ public class PulsarFunctionTlsTest {
}
}
+ private static boolean isLeaderNotReady(Throwable e) {
+ return e instanceof PulsarAdminException
+ && ((PulsarAdminException) e).getStatusCode() == 503
+ && String.valueOf(((PulsarAdminException) e).getHttpError())
+ .contains("Leader not yet ready");
+ }
+
+
protected static FunctionConfig createFunctionConfig(
String jarFile,
String tenant,
(pulsar) 02/04: [fix][fn] Fix Go function runtime to continue after user exceptions and add neg-ack tests (#25867)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 5991a1d685e4057b9c41c3973afbb7c568f188eb
Author: Dream95 <86...@qq.com>
AuthorDate: Mon May 25 20:28:29 2026 +0800
[fix][fn] Fix Go function runtime to continue after user exceptions and add neg-ack tests (#25867)
Signed-off-by: Dream95 <zh...@163.com>
(cherry picked from commit a5c1029668e182b05579d223683d5e6ea82f84fe)
---
pulsar-function-go/pf/instance.go | 41 +++++++++++++++-------
pulsar-function-go/pf/instance_test.go | 32 +++++++++++++++++
.../docker-images/latest-version-image/Dockerfile | 1 +
.../go-examples/exceptionFunc/exceptionFunc.go | 41 ++++++++++++++++++++++
.../integration/functions/PulsarFunctionsTest.java | 4 +++
.../functions/PulsarFunctionsTestBase.java | 1 +
.../functions/go/PulsarFunctionsGoTest.java | 5 +++
7 files changed, 112 insertions(+), 13 deletions(-)
diff --git a/pulsar-function-go/pf/instance.go b/pulsar-function-go/pf/instance.go
index af8a4e0157b..2cdfc8a6e94 100644
--- a/pulsar-function-go/pf/instance.go
+++ b/pulsar-function-go/pf/instance.go
@@ -164,7 +164,6 @@ CLOSE:
case cm := <-channel:
msgInput := cm.Message
atMostOnce := gi.context.instanceConf.funcDetails.ProcessingGuarantees == pb.ProcessingGuarantees_ATMOST_ONCE
- atLeastOnce := gi.context.instanceConf.funcDetails.ProcessingGuarantees == pb.ProcessingGuarantees_ATLEAST_ONCE
autoAck := gi.context.instanceConf.funcDetails.AutoAck //nolint:staticcheck
if autoAck && atMostOnce {
gi.ackInputMessage(msgInput)
@@ -177,12 +176,8 @@ CLOSE:
output, err := gi.handlerMsg(msgInput)
if err != nil {
- log.Errorf("handler message error:%v", err)
- if autoAck && atLeastOnce {
- gi.nackInputMessage(msgInput)
- }
- gi.stats.incrTotalUserExceptions(err)
- return err
+ gi.handleUserError(msgInput, err)
+ continue
}
gi.stats.processTimeEnd()
@@ -391,6 +386,29 @@ func (gi *goInstance) setupConsumer() (chan pulsar.ConsumerMessage, error) {
return channel, nil
}
+func (gi *goInstance) shouldNackInputOnFailure() bool {
+ guarantee := gi.context.instanceConf.funcDetails.ProcessingGuarantees
+ return guarantee == pb.ProcessingGuarantees_ATLEAST_ONCE ||
+ guarantee == pb.ProcessingGuarantees_MANUAL
+}
+
+func (gi *goInstance) handleUserError(msgInput pulsar.Message, err error) {
+ log.Errorf("handler message error:%v", err)
+ if gi.shouldNackInputOnFailure() {
+ gi.nackInputMessage(msgInput)
+ }
+ gi.stats.incrTotalUserExceptions(err)
+ gi.stats.processTimeEnd()
+}
+
+func (gi *goInstance) handlePublishError(msgInput pulsar.Message, err error) {
+ if gi.context.instanceConf.funcDetails.ProcessingGuarantees == pb.ProcessingGuarantees_ATLEAST_ONCE {
+ gi.nackInputMessage(msgInput)
+ }
+ gi.stats.incrTotalSysExceptions(err)
+ log.Errorf("failed to publish output message: %v", err)
+}
+
func (gi *goInstance) handlerMsg(input pulsar.Message) (output []byte, err error) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
@@ -420,11 +438,8 @@ func (gi *goInstance) processResult(msgInput pulsar.Message, output []byte) {
// semantics, ensure we nack so someone else can get it, in case we are the only handler. Then mark
// exception and fail out.
if err != nil {
- if autoAck && atLeastOnce {
- gi.nackInputMessage(msgInput)
- }
- gi.stats.incrTotalSysExceptions(err)
- log.Fatal(err)
+ gi.handlePublishError(msgInput, err)
+ return
}
// Otherwise the message succeeded. If the SDK is entrusted with responding and we are using
// atLeastOnce delivery semantics, ack the message.
@@ -437,7 +452,7 @@ func (gi *goInstance) processResult(msgInput pulsar.Message, output []byte) {
return
}
- // No output from the function or no output topic. Ack if we need to and mark the success before rturning.
+ // No output from the function or no output topic. Ack if we need to and mark the success before returning.
if autoAck && atLeastOnce {
gi.ackInputMessage(msgInput)
}
diff --git a/pulsar-function-go/pf/instance_test.go b/pulsar-function-go/pf/instance_test.go
index bf45ae3a891..447d3a22f8d 100644
--- a/pulsar-function-go/pf/instance_test.go
+++ b/pulsar-function-go/pf/instance_test.go
@@ -27,6 +27,8 @@ import (
"time"
"github.com/stretchr/testify/assert"
+
+ pb "github.com/apache/pulsar/pulsar-function-go/pb"
)
func testProcessSpawnerHealthCheckTimer(
@@ -115,3 +117,33 @@ func Test_goInstance_handlerMsg(t *testing.T) {
assert.Equal(t, "output", string(output))
assert.Equal(t, message, fc.record)
}
+
+func newTestGoInstance(guarantee pb.ProcessingGuarantees) *goInstance {
+ return &goInstance{
+ context: &FunctionContext{
+ instanceConf: &instanceConf{
+ funcDetails: pb.FunctionDetails{
+ ProcessingGuarantees: guarantee,
+ },
+ },
+ },
+ }
+}
+
+func TestShouldNackInputOnFailure(t *testing.T) {
+ tests := []struct {
+ name string
+ guarantee pb.ProcessingGuarantees
+ want bool
+ }{
+ {"atLeastOnce", pb.ProcessingGuarantees_ATLEAST_ONCE, true},
+ {"manual", pb.ProcessingGuarantees_MANUAL, true},
+ {"atMostOnce", pb.ProcessingGuarantees_ATMOST_ONCE, false},
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ instance := newTestGoInstance(tt.guarantee)
+ assert.Equal(t, tt.want, instance.shouldNackInputOnFailure())
+ })
+ }
+}
diff --git a/tests/docker-images/latest-version-image/Dockerfile b/tests/docker-images/latest-version-image/Dockerfile
index 6fe2ef656a4..08587af16c7 100644
--- a/tests/docker-images/latest-version-image/Dockerfile
+++ b/tests/docker-images/latest-version-image/Dockerfile
@@ -24,6 +24,7 @@ ARG GOLANG_IMAGE
FROM $GOLANG_IMAGE as pulsar-function-go
COPY target/pulsar-function-go/ /go/src/github.com/apache/pulsar/pulsar-function-go
+COPY go-examples/exceptionFunc/ /go/src/github.com/apache/pulsar/pulsar-function-go/examples/exceptionFunc/
RUN cd /go/src/github.com/apache/pulsar/pulsar-function-go && go install ./...
RUN cd /go/src/github.com/apache/pulsar/pulsar-function-go/pf && go install
RUN cd /go/src/github.com/apache/pulsar/pulsar-function-go/examples && go install ./...
diff --git a/tests/docker-images/latest-version-image/go-examples/exceptionFunc/exceptionFunc.go b/tests/docker-images/latest-version-image/go-examples/exceptionFunc/exceptionFunc.go
new file mode 100644
index 00000000000..80ace6e21b9
--- /dev/null
+++ b/tests/docker-images/latest-version-image/go-examples/exceptionFunc/exceptionFunc.go
@@ -0,0 +1,41 @@
+//
+// 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 main
+
+import (
+ "context"
+ "errors"
+
+ "github.com/apache/pulsar/pulsar-function-go/pf"
+)
+
+var i int
+
+func HandleException(ctx context.Context, in []byte) ([]byte, error) {
+ i++
+ if i%10 == 0 {
+ return nil, errors.New("test")
+ }
+ return []byte(string(in) + "!"), nil
+}
+
+func main() {
+ pf.Start(HandleException)
+}
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java
index fdda3144e8b..82418d9d9dc 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java
@@ -404,6 +404,10 @@ public abstract class PulsarFunctionsTest extends PulsarFunctionsTestBase {
submitFunction(
runtime, inputTopicName, outputTopicName, functionName, EXCEPTION_FUNCTION_PYTHON_FILE,
EXCEPTION_PYTHON_CLASS, schema, null);
+ } else if (runtime == Runtime.GO) {
+ submitFunction(
+ runtime, inputTopicName, outputTopicName, functionName, EXCEPTION_GO_FILE,
+ null, schema, null);
} else {
submitFunction(
runtime, inputTopicName, outputTopicName, functionName, null, EXCEPTION_JAVA_CLASS, schema, null);
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTestBase.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTestBase.java
index 288ced63ae5..84b31dca9a0 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTestBase.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTestBase.java
@@ -78,6 +78,7 @@ public abstract class PulsarFunctionsTestBase extends PulsarTestSuite {
public static final String EXCLAMATION_GO_FILE = "exclamationFunc";
public static final String PUBLISH_FUNCTION_GO_FILE = "exclamationFunc";
+ public static final String EXCEPTION_GO_FILE = "exceptionFunc";
public static final String LOGGING_JAVA_CLASS =
"org.apache.pulsar.functions.api.examples.LoggingFunction";
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/go/PulsarFunctionsGoTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/go/PulsarFunctionsGoTest.java
index 0550fd94ebe..6e631adafbf 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/go/PulsarFunctionsGoTest.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/go/PulsarFunctionsGoTest.java
@@ -39,4 +39,9 @@ public abstract class PulsarFunctionsGoTest extends PulsarFunctionsTest {
testExclamationFunction(Runtime.GO, false, false, true, false);
}
+ @Test(groups = {"go_function", "function"})
+ public void testGoFunctionNegAck() throws Exception {
+ testFunctionNegAck(Runtime.GO);
+ }
+
}
(pulsar) 03/04: [fix][test] Fix flaky ProducerCleanupTest timer cleanup (#25864)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 5f10bf0c96fa98e619af599ae208bc5694f1c789
Author: Oneby Wang <44...@users.noreply.github.com>
AuthorDate: Mon May 25 20:28:47 2026 +0800
[fix][test] Fix flaky ProducerCleanupTest timer cleanup (#25864)
(cherry picked from commit 2e02b7830ee14e8d3476dba03f7fa6fd0da175b8)
---
.../apache/pulsar/client/api/ProducerCleanupTest.java | 17 +++++++++++------
1 file changed, 11 insertions(+), 6 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerCleanupTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerCleanupTest.java
index 2a7fcb3eb17..09c0113c0a1 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerCleanupTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerCleanupTest.java
@@ -20,8 +20,10 @@ package org.apache.pulsar.client.api;
import io.netty.util.HashedWheelTimer;
import java.util.concurrent.TimeUnit;
+import lombok.Cleanup;
import org.apache.pulsar.broker.service.SharedPulsarBaseTest;
import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.awaitility.Awaitility;
import org.testng.Assert;
import org.testng.annotations.Test;
@@ -29,14 +31,17 @@ import org.testng.annotations.Test;
public class ProducerCleanupTest extends SharedPulsarBaseTest {
@Test
- public void testAllTimerTaskShouldCanceledAfterProducerClosed() throws PulsarClientException, InterruptedException {
- Producer<byte[]> producer = pulsarClient.newProducer()
+ public void testAllTimerTaskShouldCanceledAfterProducerClosed() throws PulsarClientException {
+ @Cleanup
+ PulsarClient client = newPulsarClient();
+ Producer<byte[]> producer = client.newProducer()
.topic(newTopicName())
- .sendTimeout(1, TimeUnit.SECONDS)
+ .sendTimeout(15, TimeUnit.SECONDS)
.create();
producer.close();
- Thread.sleep(2000);
- HashedWheelTimer timer = (HashedWheelTimer) ((PulsarClientImpl) pulsarClient).timer();
- Assert.assertEquals(timer.pendingTimeouts(), 0);
+ HashedWheelTimer timer = (HashedWheelTimer) ((PulsarClientImpl) client).timer();
+ Awaitility.await()
+ .atMost(10, TimeUnit.SECONDS)
+ .untilAsserted(() -> Assert.assertEquals(timer.pendingTimeouts(), 0));
}
}