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/05/22 10:35:06 UTC
(pulsar) branch branch-4.0 updated (2e79ee3364f -> c05d2085fea)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 2e79ee3364f [fix][sec] Bump org.asynchttpclient:async-http-client from 2.14.5 to 2.15.0 (#25818)
new ad7589f53b1 [fix][broker] Fix ManagedLedgerImpl.advanceCursorsIfNecessary() method may lose non-durable cursor properties in race condition (#25796)
new 3726698ad89 [improve][fn] Avoid gRPC timeout when getting status of a dead process runtime (#25819)
new c05d2085fea [fix][fn] Fix functions update issue where artifact is provided as a http url (#25840)
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:
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 2 +-
.../bookkeeper/mledger/impl/ManagedLedgerTest.java | 62 ++++++++++++++++++++++
.../functions/runtime/process/ProcessRuntime.java | 9 ++++
.../functions/worker/rest/api/ComponentImpl.java | 4 +-
.../api/v3/AbstractFunctionApiResourceTest.java | 39 ++++++++++++++
5 files changed, 113 insertions(+), 3 deletions(-)
(pulsar) 01/03: [fix][broker] Fix ManagedLedgerImpl.advanceCursorsIfNecessary() method may lose non-durable cursor properties in race condition (#25796)
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.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit ad7589f53b1b73c52f1a6b65c3f086b9b012d580
Author: Oneby Wang <44...@users.noreply.github.com>
AuthorDate: Mon May 18 17:14:10 2026 +0800
[fix][broker] Fix ManagedLedgerImpl.advanceCursorsIfNecessary() method may lose non-durable cursor properties in race condition (#25796)
(cherry picked from commit cbe716121068357de367b20e9b6f5bf08e869cb4)
---
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 2 +-
.../bookkeeper/mledger/impl/ManagedLedgerTest.java | 62 ++++++++++++++++++++++
2 files changed, 63 insertions(+), 1 deletion(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
index 95ba53312b4..f1eaa04d313 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
@@ -3149,7 +3149,7 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
&& highestPositionToDelete.compareTo(cursor.getManagedLedger()
.getLastConfirmedEntry()) <= 0 && !(!cursor.isDurable() && cursor instanceof NonDurableCursorImpl
&& ((NonDurableCursorImpl) cursor).isReadCompacted())) {
- cursor.asyncMarkDelete(highestPositionToDelete, cursor.getProperties(), new MarkDeleteCallback() {
+ cursor.asyncMarkDelete(highestPositionToDelete, null, new MarkDeleteCallback() {
@Override
public void markDeleteComplete(Object ctx) {
}
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 3804ff7ade9..dff78eab2f3 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
@@ -26,6 +26,7 @@ import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyMap;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.nullable;
import static org.mockito.Mockito.atLeast;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doNothing;
@@ -5213,4 +5214,65 @@ public class ManagedLedgerTest extends MockedBookKeeperTestCase {
// Verify properties are preserved after cursor reset
assertEquals(cursor.getProperties(), expectedProperties);
}
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testAdvanceCursorsIfNecessaryNeverLoseMarkDeleteProperties() throws Exception {
+ ManagedLedgerConfig config = new ManagedLedgerConfig();
+ config.setMaxEntriesPerLedger(1);
+ config.setRetentionTime(0, TimeUnit.SECONDS);
+ config.setRetentionSizeInMB(0);
+
+ @Cleanup
+ ManagedLedgerImpl ledger =
+ (ManagedLedgerImpl) factory.open("testAdvanceCursorsIfNecessaryNeverLoseMarkDeleteProperties", config);
+ @Cleanup
+ ManagedCursorImpl durableCursor = (ManagedCursorImpl) ledger.openCursor("durableCursor1");
+ @Cleanup
+ NonDurableCursorImpl realNonDurableCursor =
+ (NonDurableCursorImpl) ledger.newNonDurableCursor(PositionFactory.EARLIEST);
+ NonDurableCursorImpl nonDurableCursor = spy(realNonDurableCursor);
+
+ ledger.getCursors().removeCursor(realNonDurableCursor.getName());
+ ledger.getCursors().add(nonDurableCursor, null);
+
+ CountDownLatch advanceCursorsMarkDeleteEnteredLatch = new CountDownLatch(1);
+ CountDownLatch nonDurableCursorsMarkDeleteCompletedLatch = new CountDownLatch(1);
+ CountDownLatch advanceCursorsMarkDeleteCompletedLatch = new CountDownLatch(1);
+
+ doAnswer(invocation -> {
+ Map<String, Long> invocationProperties = invocation.getArgument(1);
+ // Pause the advanceCursorsIfNecessary mark-delete so the nonDurableCursor markDelete() can complete first.
+ if (invocationProperties == null || invocationProperties.isEmpty()) {
+ advanceCursorsMarkDeleteEnteredLatch.countDown();
+ assertTrue(nonDurableCursorsMarkDeleteCompletedLatch.await(5, TimeUnit.SECONDS));
+ try {
+ return invocation.callRealMethod();
+ } finally {
+ advanceCursorsMarkDeleteCompletedLatch.countDown();
+ }
+ }
+
+ return invocation.callRealMethod();
+ }).when(nonDurableCursor)
+ .internalAsyncMarkDelete(any(Position.class), nullable(Map.class), any(MarkDeleteCallback.class),
+ nullable(Object.class), nullable(Runnable.class));
+
+ ledger.addEntry("entry-1".getBytes(Encoding));
+ Position pos2 = ledger.addEntry("entry-2".getBytes(Encoding));
+
+ // Mark-delete the durable cursor to trigger trimming, which advances non-durable cursors.
+ durableCursor.markDelete(pos2);
+ assertTrue(advanceCursorsMarkDeleteEnteredLatch.await(5, TimeUnit.SECONDS));
+
+ String propertyKey = "test-property";
+ Map<String, Long> properties = new HashMap<>();
+ properties.put(propertyKey, 1L);
+ nonDurableCursor.markDelete(pos2, properties);
+ nonDurableCursorsMarkDeleteCompletedLatch.countDown();
+
+ assertTrue(advanceCursorsMarkDeleteCompletedLatch.await(5, TimeUnit.SECONDS));
+ assertEquals(nonDurableCursor.getMarkDeletedPosition(), pos2);
+ assertEquals(nonDurableCursor.getProperties(), properties);
+ }
}
(pulsar) 02/03: [improve][fn] Avoid gRPC timeout when getting status of a dead process runtime (#25819)
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.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 3726698ad89e2811332e78274c154ad16db7433c
Author: Dream95 <86...@qq.com>
AuthorDate: Tue May 19 19:43:54 2026 +0800
[improve][fn] Avoid gRPC timeout when getting status of a dead process runtime (#25819)
Signed-off-by: Dream95 <zh...@163.com>
(cherry picked from commit 5fd7518bd5522a0084d876c745a0ec870f1f2316)
---
.../apache/pulsar/functions/runtime/process/ProcessRuntime.java | 9 +++++++++
1 file changed, 9 insertions(+)
diff --git a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/process/ProcessRuntime.java b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/process/ProcessRuntime.java
index e59fcadc729..24c41431c38 100644
--- a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/process/ProcessRuntime.java
+++ b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/process/ProcessRuntime.java
@@ -242,6 +242,15 @@ class ProcessRuntime implements Runtime {
retval.completeExceptionally(new RuntimeException("Not alive"));
return retval;
}
+ if (!isAlive()) {
+ FunctionStatus.Builder builder = FunctionStatus.newBuilder();
+ builder.setRunning(false);
+ if (deathException != null && deathException.getMessage() != null) {
+ builder.setFailureException(deathException.getMessage());
+ }
+ retval.complete(builder.build());
+ return retval;
+ }
ListenableFuture<FunctionStatus> response = stub.withDeadlineAfter(GRPC_TIMEOUT_SECS, TimeUnit.SECONDS)
.getFunctionStatus(Empty.newBuilder().build());
Futures.addCallback(response, new FutureCallback<FunctionStatus>() {
(pulsar) 03/03: [fix][fn] Fix functions update issue where artifact is provided as a http url (#25840)
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.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit c05d2085fea5b7e8aaf76b4bb30e9176f3f4e447
Author: sbourkeostk <58...@users.noreply.github.com>
AuthorDate: Thu May 21 10:25:36 2026 +0100
[fix][fn] Fix functions update issue where artifact is provided as a http url (#25840)
Co-authored-by: Stephen Bourke <sb...@beyond.com>
Co-authored-by: Lari Hotari <lh...@users.noreply.github.com>
Co-authored-by: Zixuan Liu <no...@gmail.com>
(cherry picked from commit 235a7a8a4ca3cb6914f965ae2407030dbd1fa128)
---
.../functions/worker/rest/api/ComponentImpl.java | 4 +--
.../api/v3/AbstractFunctionApiResourceTest.java | 39 ++++++++++++++++++++++
2 files changed, 41 insertions(+), 2 deletions(-)
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/ComponentImpl.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/ComponentImpl.java
index 4058ebb667c..f3bebd78de6 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/ComponentImpl.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/ComponentImpl.java
@@ -1776,7 +1776,7 @@ public abstract class ComponentImpl implements Component<PulsarWorkerService> {
if (isNotBlank(functionPkgUrl)) {
componentPackageFile = getPackageFile(componentType, functionPkgUrl);
} else if (existingPackagePath.startsWith(Utils.FILE) || existingPackagePath.startsWith(Utils.HTTP)) {
- if (!worker().getPackageUrlValidator().isValidPackageUrl(componentType, functionPkgUrl)) {
+ if (!worker().getPackageUrlValidator().isValidPackageUrl(componentType, existingPackagePath)) {
throw new IllegalArgumentException("Function Package url is not valid."
+ "supported url (http/https/file)");
}
@@ -1785,7 +1785,7 @@ public abstract class ComponentImpl implements Component<PulsarWorkerService> {
} catch (Exception e) {
throw new IllegalArgumentException(String.format("Encountered error \"%s\" "
+ "when getting %s package from %s", e.getMessage(),
- ComponentTypeUtils.toString(componentType), functionPkgUrl));
+ ComponentTypeUtils.toString(componentType), existingPackagePath));
}
} else if (Utils.hasPackageTypePrefix(existingPackagePath)) {
componentPackageFile = getPackageFile(componentType, existingPackagePath);
diff --git a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/AbstractFunctionApiResourceTest.java b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/AbstractFunctionApiResourceTest.java
index d7c5b877f63..16cf98b9847 100644
--- a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/AbstractFunctionApiResourceTest.java
+++ b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/AbstractFunctionApiResourceTest.java
@@ -1028,6 +1028,45 @@ public abstract class AbstractFunctionApiResourceTest extends AbstractFunctionsR
}
+ @Test
+ public void testUpdateFunctionWithExistingFileUrl() throws IOException {
+
+ String fileLocation = FutureUtil.class.getProtectionDomain().getCodeSource().getLocation().getPath();
+ String filePackageUrl = "file://" + fileLocation;
+
+ FunctionConfig functionConfig = new FunctionConfig();
+ functionConfig.setOutput(OUTPUT_TOPIC);
+ functionConfig.setOutputSerdeClassName(OUTPUT_SERDE_CLASS_NAME);
+ functionConfig.setTenant(TENANT);
+ functionConfig.setNamespace(NAMESPACE);
+ functionConfig.setName(FUNCTION);
+ functionConfig.setClassName(CLASS_NAME);
+ // increment parallelism to avoid 'Update contains no change' exception
+ functionConfig.setParallelism(PARALLELISM + 1);
+ functionConfig.setRuntime(FunctionConfig.Runtime.JAVA);
+ functionConfig.setCustomSerdeInputs(TOPICS_TO_SER_DE_CLASS_NAME);
+
+ FunctionMetaData existingMetaData = FunctionMetaData.newBuilder()
+ .setFunctionDetails(createDefaultFunctionDetails())
+ .setPackageLocation(org.apache.pulsar.functions.proto.Function.PackageLocationMetaData.newBuilder()
+ .setPackagePath(filePackageUrl)
+ .build())
+ .build();
+
+ when(mockedManager.containsFunction(eq(TENANT), eq(NAMESPACE), eq(FUNCTION))).thenReturn(true);
+ when(mockedManager.getFunctionMetaData(any(), any(), any())).thenReturn(existingMetaData);
+
+ updateFunction(
+ TENANT,
+ NAMESPACE,
+ FUNCTION,
+ null,
+ null,
+ null,
+ functionConfig,
+ null, null);
+ }
+
@Test(expectedExceptions = RestException.class, expectedExceptionsMessageRegExp = "function failed to register")
public void testUpdateFunctionFailure() throws Exception {
try {