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 2024/12/18 15:07:19 UTC
(pulsar) branch branch-3.3 updated (0c6ce59a84f -> 593625cebee)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 0c6ce59a84f [improve] Upgrade lombok to 1.18.36 (#23752)
new 789b9ccf7a0 [improve][fn] Improve implementation for maxPendingAsyncRequests async concurrency limit when return type is CompletableFuture<Void> (#23708)
new 593625cebee [improve][fn] Improve closing of producers in Pulsar Functions ProducerCache invalidation (#23734)
The 2 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:
.../functions/instance/JavaExecutionResult.java | 4 +-
.../pulsar/functions/instance/JavaInstance.java | 77 +++++++++++++++++-----
.../functions/instance/JavaInstanceRunnable.java | 2 +-
.../pulsar/functions/instance/ProducerCache.java | 38 ++++++++---
.../functions/instance/JavaInstanceTest.java | 64 ++++++++++++++++++
.../functions/instance/ProducerCacheTest.java | 64 ++++++++++++++++++
6 files changed, 219 insertions(+), 30 deletions(-)
create mode 100644 pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/ProducerCacheTest.java
(pulsar) 01/02: [improve][fn] Improve implementation for maxPendingAsyncRequests async concurrency limit when return type is CompletableFuture (#23708)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 789b9ccf7a0afdf1d9fdba1bc197f36476aa687c
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Tue Dec 17 18:33:25 2024 +0200
[improve][fn] Improve implementation for maxPendingAsyncRequests async concurrency limit when return type is CompletableFuture<Void> (#23708)
(cherry picked from commit 8ad67776fc0787fe857d999a72f9df1e95e4210d)
---
.../functions/instance/JavaExecutionResult.java | 4 +-
.../pulsar/functions/instance/JavaInstance.java | 77 +++++++++++++++++-----
.../functions/instance/JavaInstanceRunnable.java | 2 +-
.../functions/instance/JavaInstanceTest.java | 64 ++++++++++++++++++
4 files changed, 127 insertions(+), 20 deletions(-)
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaExecutionResult.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaExecutionResult.java
index 7af238154d6..5856600196b 100644
--- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaExecutionResult.java
+++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaExecutionResult.java
@@ -27,13 +27,11 @@ import lombok.Data;
*/
@Data
public class JavaExecutionResult {
- private Exception userException;
- private Exception systemException;
+ private Throwable userException;
private Object result;
public void reset() {
setUserException(null);
- setSystemException(null);
setResult(null);
}
}
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstance.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstance.java
index 292f52b5091..5946be9fe5b 100644
--- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstance.java
+++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstance.java
@@ -25,11 +25,13 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.Semaphore;
import java.util.function.Consumer;
import lombok.AccessLevel;
import lombok.Data;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.functions.api.Function;
import org.apache.pulsar.functions.api.Record;
@@ -57,13 +59,26 @@ public class JavaInstance implements AutoCloseable {
private final ExecutorService executor;
@Getter
private final LinkedBlockingQueue<AsyncFuncRequest> pendingAsyncRequests;
+ @Getter
+ private final Semaphore asyncRequestsConcurrencyLimiter;
+ private final boolean asyncPreserveInputOrderForOutputMessages;
public JavaInstance(ContextImpl contextImpl, Object userClassObject, InstanceConfig instanceConfig) {
this.context = contextImpl;
this.instanceConfig = instanceConfig;
this.executor = Executors.newSingleThreadExecutor();
- this.pendingAsyncRequests = new LinkedBlockingQueue<>(this.instanceConfig.getMaxPendingAsyncRequests());
+
+ asyncPreserveInputOrderForOutputMessages =
+ resolveAsyncPreserveInputOrderForOutputMessages(instanceConfig);
+
+ if (asyncPreserveInputOrderForOutputMessages) {
+ this.pendingAsyncRequests = new LinkedBlockingQueue<>(this.instanceConfig.getMaxPendingAsyncRequests());
+ this.asyncRequestsConcurrencyLimiter = null;
+ } else {
+ this.pendingAsyncRequests = null;
+ this.asyncRequestsConcurrencyLimiter = new Semaphore(this.instanceConfig.getMaxPendingAsyncRequests());
+ }
// create the functions
if (userClassObject instanceof Function) {
@@ -73,6 +88,20 @@ public class JavaInstance implements AutoCloseable {
}
}
+ // resolve whether to preserve input order for output messages for async functions
+ private boolean resolveAsyncPreserveInputOrderForOutputMessages(InstanceConfig instanceConfig) {
+ // no need to preserve input order for output messages if the function returns Void type
+ boolean voidReturnType = instanceConfig.getFunctionDetails() != null
+ && instanceConfig.getFunctionDetails().getSink() != null
+ && Void.class.getName().equals(instanceConfig.getFunctionDetails().getSink().getTypeClassName());
+ if (voidReturnType) {
+ return false;
+ }
+
+ // preserve input order for output messages
+ return true;
+ }
+
@VisibleForTesting
public JavaExecutionResult handleMessage(Record<?> record, Object input) {
return handleMessage(record, input, (rec, result) -> {
@@ -103,15 +132,33 @@ public class JavaInstance implements AutoCloseable {
}
if (output instanceof CompletableFuture) {
- // Function is in format: Function<I, CompletableFuture<O>>
- AsyncFuncRequest request = new AsyncFuncRequest(
- record, (CompletableFuture) output
- );
try {
- pendingAsyncRequests.put(request);
- ((CompletableFuture) output).whenCompleteAsync((res, cause) -> {
+ if (asyncPreserveInputOrderForOutputMessages) {
+ // Function is in format: Function<I, CompletableFuture<O>>
+ AsyncFuncRequest request = new AsyncFuncRequest(
+ record, (CompletableFuture) output
+ );
+ pendingAsyncRequests.put(request);
+ } else {
+ asyncRequestsConcurrencyLimiter.acquire();
+ }
+ ((CompletableFuture<Object>) output).whenCompleteAsync((Object res, Throwable cause) -> {
try {
- processAsyncResults(asyncResultConsumer);
+ if (asyncPreserveInputOrderForOutputMessages) {
+ processAsyncResultsInInputOrder(asyncResultConsumer);
+ } else {
+ try {
+ JavaExecutionResult execResult = new JavaExecutionResult();
+ if (cause != null) {
+ execResult.setUserException(FutureUtil.unwrapCompletionException(cause));
+ } else {
+ execResult.setResult(res);
+ }
+ asyncResultConsumer.accept(record, execResult);
+ } finally {
+ asyncRequestsConcurrencyLimiter.release();
+ }
+ }
} catch (Throwable innerException) {
// the thread used for processing async results failed
asyncFailureHandler.accept(innerException);
@@ -132,21 +179,20 @@ public class JavaInstance implements AutoCloseable {
}
}
- private void processAsyncResults(JavaInstanceRunnable.AsyncResultConsumer resultConsumer) throws Exception {
+ // processes the async results in the input order so that the order of the result messages in the output topic
+ // are in the same order as the input
+ private void processAsyncResultsInInputOrder(JavaInstanceRunnable.AsyncResultConsumer resultConsumer)
+ throws Exception {
AsyncFuncRequest asyncResult = pendingAsyncRequests.peek();
while (asyncResult != null && asyncResult.getProcessResult().isDone()) {
pendingAsyncRequests.remove(asyncResult);
- JavaExecutionResult execResult = new JavaExecutionResult();
+ JavaExecutionResult execResult = new JavaExecutionResult();
try {
Object result = asyncResult.getProcessResult().get();
execResult.setResult(result);
} catch (ExecutionException e) {
- if (e.getCause() instanceof Exception) {
- execResult.setUserException((Exception) e.getCause());
- } else {
- execResult.setUserException(new Exception(e.getCause()));
- }
+ execResult.setUserException(FutureUtil.unwrapCompletionException(e));
}
resultConsumer.accept(asyncResult.getRecord(), execResult);
@@ -154,7 +200,6 @@ public class JavaInstance implements AutoCloseable {
// peek the next result
asyncResult = pendingAsyncRequests.peek();
}
-
}
public void initialize() throws Exception {
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
index baf0c5f7400..4f811c14704 100644
--- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
+++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
@@ -423,7 +423,7 @@ public class JavaInstanceRunnable implements AutoCloseable, Runnable {
@VisibleForTesting
void handleResult(Record srcRecord, JavaExecutionResult result) throws Exception {
if (result.getUserException() != null) {
- Exception t = result.getUserException();
+ Throwable t = result.getUserException();
log.warn("Encountered exception when processing message {}",
srcRecord, t);
stats.incrUserExceptions(t);
diff --git a/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceTest.java b/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceTest.java
index 5a333204293..b3fcef292e5 100644
--- a/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceTest.java
+++ b/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceTest.java
@@ -23,10 +23,13 @@ import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertSame;
+import static org.testng.Assert.assertTrue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.functions.api.Function;
@@ -245,4 +248,65 @@ public class JavaInstanceTest {
super(msg);
}
}
+
+ @Test
+ public void testAsyncFunctionMaxPendingVoidResult() throws Exception {
+ CountDownLatch count = new CountDownLatch(1);
+ InstanceConfig instanceConfig = new InstanceConfig();
+ instanceConfig.setFunctionDetails(org.apache.pulsar.functions.proto.Function.FunctionDetails.newBuilder()
+ .setSink(org.apache.pulsar.functions.proto.Function.SinkSpec.newBuilder()
+ .setTypeClassName(Void.class.getName())
+ .build())
+ .build());
+ int pendingQueueSize = 3;
+ instanceConfig.setMaxPendingAsyncRequests(pendingQueueSize);
+ @Cleanup("shutdownNow")
+ ExecutorService executor = Executors.newCachedThreadPool();
+
+ Function<String, CompletableFuture<Void>> function = (input, context) -> {
+ CompletableFuture<Void> result = new CompletableFuture<>();
+ executor.submit(() -> {
+ try {
+ count.await();
+ result.complete(null);
+ } catch (Exception e) {
+ result.completeExceptionally(e);
+ }
+ });
+
+ return result;
+ };
+
+ JavaInstance instance = new JavaInstance(
+ mock(ContextImpl.class),
+ function,
+ instanceConfig);
+ String testString = "ABC123";
+
+ CountDownLatch resultsLatch = new CountDownLatch(3);
+
+ long startTime = System.currentTimeMillis();
+ assertEquals(pendingQueueSize, instance.getAsyncRequestsConcurrencyLimiter().availablePermits());
+ JavaInstanceRunnable.AsyncResultConsumer asyncResultConsumer = (rec, result) -> {
+ resultsLatch.countDown();
+ };
+ Consumer<Throwable> asyncFailureHandler = cause -> {
+ };
+ assertNull(instance.handleMessage(mock(Record.class), testString, asyncResultConsumer, asyncFailureHandler));
+ assertEquals(pendingQueueSize - 1, instance.getAsyncRequestsConcurrencyLimiter().availablePermits());
+ assertNull(instance.handleMessage(mock(Record.class), testString, asyncResultConsumer, asyncFailureHandler));
+ assertEquals(pendingQueueSize - 2, instance.getAsyncRequestsConcurrencyLimiter().availablePermits());
+ assertNull(instance.handleMessage(mock(Record.class), testString, asyncResultConsumer, asyncFailureHandler));
+ // no space left
+ assertEquals(0, instance.getAsyncRequestsConcurrencyLimiter().availablePermits());
+
+ count.countDown();
+
+ assertTrue(resultsLatch.await(5, TimeUnit.SECONDS));
+
+ long endTime = System.currentTimeMillis();
+
+ log.info("start:{} end:{} during:{}", startTime, endTime, endTime - startTime);
+ instance.close();
+ }
}
(pulsar) 02/02: [improve][fn] Improve closing of producers in Pulsar Functions ProducerCache invalidation (#23734)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 593625cebee5f9bf2042317362a803830987d321
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Tue Dec 17 18:34:32 2024 +0200
[improve][fn] Improve closing of producers in Pulsar Functions ProducerCache invalidation (#23734)
(cherry picked from commit 9f046a5f6bfa35bc89e74635cbf7aacd43bc9fc9)
---
.../pulsar/functions/instance/ProducerCache.java | 38 +++++++++----
.../functions/instance/ProducerCacheTest.java | 64 ++++++++++++++++++++++
2 files changed, 92 insertions(+), 10 deletions(-)
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ProducerCache.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ProducerCache.java
index f68c4e95895..2e10581b352 100644
--- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ProducerCache.java
+++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ProducerCache.java
@@ -32,6 +32,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.common.util.FutureUtil;
@Slf4j
@@ -61,24 +62,41 @@ public class ProducerCache implements Closeable {
private final CopyOnWriteArrayList<CompletableFuture<Void>> closeFutures = new CopyOnWriteArrayList<>();
public ProducerCache() {
- Caffeine<ProducerCacheKey, Producer> builder = Caffeine.newBuilder()
+ Caffeine<ProducerCacheKey, Producer<?>> builder = Caffeine.newBuilder()
.scheduler(Scheduler.systemScheduler())
- .<ProducerCacheKey, Producer>removalListener((key, producer, cause) -> {
+ .<ProducerCacheKey, Producer<?>>removalListener((key, producer, cause) -> {
log.info("Closing producer for topic {}, cause {}", key.topic(), cause);
CompletableFuture closeFuture =
- producer.flushAsync()
+ CompletableFuture.supplyAsync(() -> producer.flushAsync(), Runnable::run)
.orTimeout(FLUSH_OR_CLOSE_TIMEOUT_SECONDS, TimeUnit.SECONDS)
.exceptionally(ex -> {
- log.error("Error flushing producer for topic {}", key.topic(), ex);
+ Throwable unwrappedCause = FutureUtil.unwrapCompletionException(ex);
+ if (unwrappedCause instanceof PulsarClientException.AlreadyClosedException) {
+ log.error(
+ "Error flushing producer for topic {} due to "
+ + "AlreadyClosedException",
+ key.topic());
+ } else {
+ log.error("Error flushing producer for topic {}", key.topic(),
+ unwrappedCause);
+ }
return null;
}).thenCompose(__ ->
producer.closeAsync().orTimeout(FLUSH_OR_CLOSE_TIMEOUT_SECONDS,
- TimeUnit.SECONDS)
- .exceptionally(ex -> {
- log.error("Error closing producer for topic {}", key.topic(),
- ex);
- return null;
- }));
+ TimeUnit.SECONDS)
+ ).exceptionally(ex -> {
+ Throwable unwrappedCause = FutureUtil.unwrapCompletionException(ex);
+ if (unwrappedCause instanceof PulsarClientException.AlreadyClosedException) {
+ log.error(
+ "Error closing producer for topic {} due to "
+ + "AlreadyClosedException",
+ key.topic());
+ } else {
+ log.error("Error closing producer for topic {}", key.topic(),
+ unwrappedCause);
+ }
+ return null;
+ });
if (closed.get()) {
closeFutures.add(closeFuture);
}
diff --git a/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/ProducerCacheTest.java b/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/ProducerCacheTest.java
new file mode 100644
index 00000000000..af95a7901b6
--- /dev/null
+++ b/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/ProducerCacheTest.java
@@ -0,0 +1,64 @@
+/*
+ * 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.functions.instance;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import java.util.concurrent.CompletableFuture;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.testng.annotations.Test;
+
+public class ProducerCacheTest {
+
+ @Test
+ public void shouldTolerateAlreadyClosedExceptionInClose() {
+ ProducerCache cache = new ProducerCache();
+ Producer producer = mock(Producer.class);
+ when(producer.flushAsync()).thenReturn(CompletableFuture.completedFuture(null));
+ when(producer.closeAsync()).thenReturn(
+ CompletableFuture.failedFuture(new PulsarClientException.AlreadyClosedException("Already closed")));
+ cache.getOrCreateProducer(ProducerCache.CacheArea.CONTEXT_CACHE, "topic", "key",
+ () -> (Producer<Object>) producer);
+ cache.close();
+ }
+
+ @Test
+ public void shouldTolerateRuntimeExceptionInClose() {
+ ProducerCache cache = new ProducerCache();
+ Producer producer = mock(Producer.class);
+ when(producer.flushAsync()).thenReturn(CompletableFuture.completedFuture(null));
+ when(producer.closeAsync()).thenThrow(new RuntimeException("Some exception"));
+ cache.getOrCreateProducer(ProducerCache.CacheArea.CONTEXT_CACHE, "topic", "key",
+ () -> (Producer<Object>) producer);
+ cache.close();
+ }
+
+ @Test
+ public void shouldTolerateRuntimeExceptionInFlush() {
+ ProducerCache cache = new ProducerCache();
+ Producer producer = mock(Producer.class);
+ when(producer.flushAsync()).thenThrow(new RuntimeException("Some exception"));
+ when(producer.closeAsync()).thenReturn(CompletableFuture.completedFuture(null));
+ cache.getOrCreateProducer(ProducerCache.CacheArea.CONTEXT_CACHE, "topic", "key",
+ () -> (Producer<Object>) producer);
+ cache.close();
+ }
+
+}
\ No newline at end of file