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/06/19 11:12:05 UTC
(pulsar) branch branch-3.2 updated (c7e373ad860 -> 05bd2d81d0f)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-3.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from c7e373ad860 [fix] Revert "[fix][cli] Fix the shell script parameter passthrough syntax (#22867)" (#22921)
new 0b3d8d52e21 [fix][misc] Topic name from persistence name should decode local name (#22879)
new b451530c088 [improve][broker] Optimize PersistentTopic.getLastDispatchablePosition (#22707)
new 20c34139750 [fix][fn] Enable optimized Netty direct byte buffer support for Pulsar Function runtimes (#22910)
new 05bd2d81d0f [fix][test] Fix TableViewBuilderImplTest NPE and infinite loop (#22924)
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:
.../broker/service/persistent/PersistentTopic.java | 66 ++++++++++++++++++----
.../buffer/impl/InMemTransactionBuffer.java | 14 ++++-
.../buffer/impl/TopicTransactionBuffer.java | 11 ++++
.../buffer/impl/TransactionBufferDisable.java | 14 ++++-
.../client/impl/TableViewBuilderImplTest.java | 33 +++++++----
.../org/apache/pulsar/common/naming/TopicName.java | 5 +-
.../apache/pulsar/common/naming/TopicNameTest.java | 6 ++
.../pulsar/functions/runtime/RuntimeUtils.java | 18 +++++-
.../runtime/kubernetes/KubernetesRuntimeTest.java | 36 ++++++------
.../runtime/process/ProcessRuntimeTest.java | 16 ++++--
10 files changed, 164 insertions(+), 55 deletions(-)
(pulsar) 02/04: [improve][broker] Optimize PersistentTopic.getLastDispatchablePosition (#22707)
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.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit b451530c088a87406eee3e7e797b8cfb60831a08
Author: 道君 <da...@apache.org>
AuthorDate: Tue Jun 11 23:45:12 2024 +0800
[improve][broker] Optimize PersistentTopic.getLastDispatchablePosition (#22707)
[PersistentTopic#getLastDispatchablePosition](https://github.com/apache/pulsar/blob/master/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java#L3776-L3788) is using by
[Reader#hasMessageAvailable](https://github.com/apache/pulsar/blob/master/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Reader.java#L116) , [ConsumerImpl#hasMessageAvailable](https://github.com/apache/pulsar/blob/master/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java#L2440-L2448), [Consumer#getLastMessageIdAsync](https://github.com/apache/pulsar/blob/master/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Consumer.java#L591-L615).
The current implementation is read entries from Bookkeeper(or sth else), which leads to low throughput, high latency and heavy load, this PR is for the purpose of optimization.
(cherry picked from commit 266243cae246a6fa52b4b6c626932885ad44cbf4)
---
.../broker/service/persistent/PersistentTopic.java | 66 ++++++++++++++++++----
.../buffer/impl/InMemTransactionBuffer.java | 14 ++++-
.../buffer/impl/TopicTransactionBuffer.java | 11 ++++
.../buffer/impl/TransactionBufferDisable.java | 14 ++++-
4 files changed, 89 insertions(+), 16 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index cc85d782631..5f951e0b32d 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -280,6 +280,9 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
private volatile CloseFutures closeFutures;
private Set<String> additionalSystemCursorNames = new TreeSet<>();
+ // The last position that can be dispatched to consumers
+ private volatile Position lastDispatchablePosition;
+
/***
* We use 3 futures to prevent a new closing if there is an in-progress deletion or closing. We make Pulsar return
* the in-progress one when it is called the second time.
@@ -3576,18 +3579,57 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
@Override
public CompletableFuture<Position> getLastDispatchablePosition() {
- return ManagedLedgerImplUtils.asyncGetLastValidPosition((ManagedLedgerImpl) ledger, entry -> {
- MessageMetadata md = Commands.parseMessageMetadata(entry.getDataBuffer());
- // If a messages has marker will filter by AbstractBaseDispatcher.filterEntriesForConsumer
- if (Markers.isServerOnlyMarker(md)) {
- return false;
- } else if (md.hasTxnidMostBits() && md.hasTxnidLeastBits()) {
- // Filter-out transaction aborted messages.
- TxnID txnID = new TxnID(md.getTxnidMostBits(), md.getTxnidLeastBits());
- return !isTxnAborted(txnID, (PositionImpl) entry.getPosition());
- }
- return true;
- }, getMaxReadPosition());
+ if (lastDispatchablePosition != null) {
+ return CompletableFuture.completedFuture(lastDispatchablePosition);
+ }
+ return ManagedLedgerImplUtils
+ .asyncGetLastValidPosition((ManagedLedgerImpl) ledger, entry -> {
+ MessageMetadata md = Commands.parseMessageMetadata(entry.getDataBuffer());
+ // If a messages has marker will filter by AbstractBaseDispatcher.filterEntriesForConsumer
+ if (Markers.isServerOnlyMarker(md)) {
+ return false;
+ } else if (md.hasTxnidMostBits() && md.hasTxnidLeastBits()) {
+ // Filter-out transaction aborted messages.
+ TxnID txnID = new TxnID(md.getTxnidMostBits(), md.getTxnidLeastBits());
+ return !isTxnAborted(txnID, (PositionImpl) entry.getPosition());
+ }
+ return true;
+ }, getMaxReadPosition())
+ .thenApply(position -> {
+ // Update lastDispatchablePosition to the given position
+ updateLastDispatchablePosition(position);
+ return position;
+ });
+ }
+
+ /**
+ * Update lastDispatchablePosition if the given position is greater than the lastDispatchablePosition.
+ *
+ * @param position
+ */
+ public synchronized void updateLastDispatchablePosition(Position position) {
+ // Update lastDispatchablePosition to null if the position is null, fallback to
+ // ManagedLedgerImplUtils#asyncGetLastValidPosition
+ if (position == null) {
+ lastDispatchablePosition = null;
+ return;
+ }
+
+ PositionImpl position0 = (PositionImpl) position;
+ // If the position is greater than the maxReadPosition, ignore
+ if (position0.compareTo(getMaxReadPosition()) > 0) {
+ return;
+ }
+ // If the lastDispatchablePosition is null, set it to the position
+ if (lastDispatchablePosition == null) {
+ lastDispatchablePosition = position;
+ return;
+ }
+ // If the position is greater than the lastDispatchablePosition, update it
+ PositionImpl lastDispatchablePosition0 = (PositionImpl) lastDispatchablePosition;
+ if (position0.compareTo(lastDispatchablePosition0) > 0) {
+ lastDispatchablePosition = position;
+ }
}
@Override
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBuffer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBuffer.java
index bab7b64c608..533d0716d41 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBuffer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBuffer.java
@@ -377,8 +377,11 @@ class InMemTransactionBuffer implements TransactionBuffer {
@Override
public void syncMaxReadPositionForNormalPublish(PositionImpl position, boolean isMarkerMessage) {
- if (!isMarkerMessage && maxReadPositionCallBack != null) {
- maxReadPositionCallBack.maxReadPositionMovedForward(null, position);
+ if (!isMarkerMessage) {
+ updateLastDispatchablePosition(position);
+ if (maxReadPositionCallBack != null) {
+ maxReadPositionCallBack.maxReadPositionMovedForward(null, position);
+ }
}
}
@@ -436,4 +439,11 @@ class InMemTransactionBuffer implements TransactionBuffer {
.filter(txnBuffer -> txnBuffer.status.equals(TxnStatus.COMMITTED))
.count();
}
+
+ // ThreadSafe
+ private void updateLastDispatchablePosition(Position position) {
+ if (topic instanceof PersistentTopic t) {
+ t.updateLastDispatchablePosition(position);
+ }
+ }
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java
index dfb73815e08..fbd4ddf7da0 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java
@@ -297,6 +297,11 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
}
}
+ // ThreadSafe
+ private void updateLastDispatchablePosition(Position position) {
+ topic.updateLastDispatchablePosition(position);
+ }
+
@Override
public CompletableFuture<TransactionBufferReader> openTransactionBufferReader(TxnID txnID, long startSequenceId) {
return null;
@@ -459,6 +464,8 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
} else {
updateMaxReadPosition((PositionImpl) topic.getManagedLedger().getLastConfirmedEntry(), false);
}
+ // Update the last dispatchable position to null if there is a TXN finished.
+ updateLastDispatchablePosition(null);
}
/**
@@ -523,6 +530,10 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
}
}
}
+ // If the message is a normal message, update the last dispatchable position.
+ if (!isMarkerMessage) {
+ updateLastDispatchablePosition(position);
+ }
}
@Override
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TransactionBufferDisable.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TransactionBufferDisable.java
index ebd61dbaa82..6f5dc0cd4d0 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TransactionBufferDisable.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TransactionBufferDisable.java
@@ -99,8 +99,11 @@ public class TransactionBufferDisable implements TransactionBuffer {
@Override
public void syncMaxReadPositionForNormalPublish(PositionImpl position, boolean isMarkerMessage) {
- if (!isMarkerMessage && maxReadPositionCallBack != null) {
- maxReadPositionCallBack.maxReadPositionMovedForward(null, position);
+ if (!isMarkerMessage) {
+ updateLastDispatchablePosition(position);
+ if (maxReadPositionCallBack != null) {
+ maxReadPositionCallBack.maxReadPositionMovedForward(null, position);
+ }
}
}
@@ -148,4 +151,11 @@ public class TransactionBufferDisable implements TransactionBuffer {
public long getCommittedTxnCount() {
return 0;
}
+
+ // ThreadSafe
+ private void updateLastDispatchablePosition(Position position) {
+ if (topic instanceof PersistentTopic t) {
+ t.updateLastDispatchablePosition(position);
+ }
+ }
}
(pulsar) 04/04: [fix][test] Fix TableViewBuilderImplTest NPE and infinite loop (#22924)
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.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 05bd2d81d0f19afeef8d7e60a3e4a29a58b86360
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Mon Jun 17 21:13:10 2024 +0300
[fix][test] Fix TableViewBuilderImplTest NPE and infinite loop (#22924)
(cherry picked from commit 2dc0d96fa0da696949414d86fb11a62beca7cb3f)
---
.../client/impl/TableViewBuilderImplTest.java | 33 ++++++++++++++--------
1 file changed, 21 insertions(+), 12 deletions(-)
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/TableViewBuilderImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/TableViewBuilderImplTest.java
index eee8ba4e8f4..01353e47cd0 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/TableViewBuilderImplTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/TableViewBuilderImplTest.java
@@ -18,6 +18,14 @@
*/
package org.apache.pulsar.client.impl;
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertNotNull;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
import org.apache.pulsar.client.api.ConsumerCryptoFailureAction;
import org.apache.pulsar.client.api.CryptoKeyReader;
import org.apache.pulsar.client.api.PulsarClientException;
@@ -25,32 +33,25 @@ import org.apache.pulsar.client.api.Reader;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.TableView;
import org.apache.pulsar.client.impl.conf.ReaderConfigurationData;
+import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.TimeUnit;
-
-import static org.mockito.Mockito.any;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-import static org.testng.Assert.assertNotNull;
-
/**
- * Unit tests of {@link TablewViewBuilderImpl}.
+ * Unit tests of {@link TableViewBuilderImpl}.
*/
public class TableViewBuilderImplTest {
private static final String TOPIC_NAME = "testTopicName";
private PulsarClientImpl client;
private TableViewBuilderImpl tableViewBuilderImpl;
+ private CompletableFuture readNextFuture;
@BeforeClass(alwaysRun = true)
public void setup() {
Reader reader = mock(Reader.class);
- when(reader.readNextAsync()).thenReturn(CompletableFuture.allOf());
+ readNextFuture = new CompletableFuture();
+ when(reader.readNextAsync()).thenReturn(readNextFuture);
client = mock(PulsarClientImpl.class);
ConnectionPool connectionPool = mock(ConnectionPool.class);
when(client.getCnxPool()).thenReturn(connectionPool);
@@ -61,6 +62,14 @@ public class TableViewBuilderImplTest {
tableViewBuilderImpl = new TableViewBuilderImpl(client, Schema.BYTES);
}
+ @AfterClass(alwaysRun = true)
+ public void cleanup() {
+ if (readNextFuture != null) {
+ readNextFuture.completeExceptionally(new PulsarClientException.AlreadyClosedException("Closing test case"));
+ readNextFuture = null;
+ }
+ }
+
@Test
public void testTableViewBuilderImpl() throws PulsarClientException {
TableView tableView = tableViewBuilderImpl.topic(TOPIC_NAME)
(pulsar) 03/04: [fix][fn] Enable optimized Netty direct byte buffer support for Pulsar Function runtimes (#22910)
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.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 20c3413975047eaefd19668732e0a4ccd7e48f85
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Mon Jun 17 09:26:28 2024 +0300
[fix][fn] Enable optimized Netty direct byte buffer support for Pulsar Function runtimes (#22910)
(cherry picked from commit f3d4d5ac0442eed2b538b8587186cdc0b8df9987)
---
.../pulsar/functions/runtime/RuntimeUtils.java | 18 +++++++++--
.../runtime/kubernetes/KubernetesRuntimeTest.java | 36 ++++++++++++----------
.../runtime/process/ProcessRuntimeTest.java | 16 ++++++----
3 files changed, 46 insertions(+), 24 deletions(-)
diff --git a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/RuntimeUtils.java b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/RuntimeUtils.java
index 0214b18fb23..78347948688 100644
--- a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/RuntimeUtils.java
+++ b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/RuntimeUtils.java
@@ -361,12 +361,26 @@ public class RuntimeUtils {
instanceConfig.getFunctionDetails().getName(),
shardId));
+ // Needed for optimized Netty direct byte buffer support
args.add("-Dio.netty.tryReflectionSetAccessible=true");
+ // Handle possible shaded Netty versions
+ args.add("-Dorg.apache.pulsar.shade.io.netty.tryReflectionSetAccessible=true");
+ args.add("-Dio.grpc.netty.shaded.io.netty.tryReflectionSetAccessible=true");
+
+ if (SystemUtils.isJavaVersionAtLeast(JavaVersion.JAVA_11)) {
+ // Needed for optimized Netty direct byte buffer support
+ args.add("--add-opens");
+ args.add("java.base/java.nio=ALL-UNNAMED");
+ args.add("--add-opens");
+ args.add("java.base/jdk.internal.misc=ALL-UNNAMED");
+ }
- // Needed for netty.DnsResolverUtil on JDK9+
if (SystemUtils.isJavaVersionAtLeast(JavaVersion.JAVA_9)) {
+ // Needed for optimized checksum calculation when com.scurrilous.circe.checksum.Java9IntHash
+ // is used. That gets used when the native library libcirce-checksum is not available or cannot
+ // be loaded.
args.add("--add-opens");
- args.add("java.base/sun.net=ALL-UNNAMED");
+ args.add("java.base/java.util.zip=ALL-UNNAMED");
}
if (instanceConfig.getAdditionalJavaRuntimeArguments() != null) {
diff --git a/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/kubernetes/KubernetesRuntimeTest.java b/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/kubernetes/KubernetesRuntimeTest.java
index 02f3c0d23fb..6ed98494129 100644
--- a/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/kubernetes/KubernetesRuntimeTest.java
+++ b/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/kubernetes/KubernetesRuntimeTest.java
@@ -441,14 +441,14 @@ public class KubernetesRuntimeTest {
if (null != depsDir) {
extraDepsEnv = " -Dpulsar.functions.extra.dependencies.dir=" + depsDir;
classpath = classpath + ":" + depsDir + "/*";
- totalArgs = 46;
- portArg = 33;
- metricsPortArg = 35;
+ totalArgs = 52;
+ portArg = 39;
+ metricsPortArg = 41;
} else {
extraDepsEnv = "";
- portArg = 32;
- metricsPortArg = 34;
- totalArgs = 45;
+ portArg = 38;
+ metricsPortArg = 40;
+ totalArgs = 51;
}
if (secretsAttached) {
totalArgs += 4;
@@ -479,7 +479,11 @@ public class KubernetesRuntimeTest {
+ "-Dpulsar.function.log.dir=" + logDirectory + "/" + FunctionCommon.getFullyQualifiedName(config.getFunctionDetails())
+ " -Dpulsar.function.log.file=" + config.getFunctionDetails().getName() + "-$SHARD_ID"
+ " -Dio.netty.tryReflectionSetAccessible=true"
- + " --add-opens java.base/sun.net=ALL-UNNAMED"
+ + " -Dorg.apache.pulsar.shade.io.netty.tryReflectionSetAccessible=true"
+ + " -Dio.grpc.netty.shaded.io.netty.tryReflectionSetAccessible=true"
+ + " --add-opens java.base/java.nio=ALL-UNNAMED"
+ + " --add-opens java.base/jdk.internal.misc=ALL-UNNAMED"
+ + " --add-opens java.base/java.util.zip=ALL-UNNAMED"
+ " -Xmx" + RESOURCES.getRam()
+ " org.apache.pulsar.functions.instance.JavaInstanceMain"
+ " --jar " + jarLocation
@@ -1306,7 +1310,7 @@ public class KubernetesRuntimeTest {
.contains("--metrics_port 0"));
}
}
-
+
@Test
public void testDeleteStatefulSetWithTranslatedKubernetesLabelChars() throws Exception {
InstanceConfig config = createJavaInstanceConfig(FunctionDetails.Runtime.JAVA, false);
@@ -1315,22 +1319,22 @@ public class KubernetesRuntimeTest {
CoreV1Api coreApi = mock(CoreV1Api.class);
AppsV1Api appsApi = mock(AppsV1Api.class);
-
+
Call successfulCall = mock(Call.class);
Response okResponse = mock(Response.class);
when(okResponse.code()).thenReturn(HttpURLConnection.HTTP_OK);
when(okResponse.isSuccessful()).thenReturn(true);
when(okResponse.message()).thenReturn("");
when(successfulCall.execute()).thenReturn(okResponse);
-
+
final String expectedFunctionNamePrefix = String.format("pf-%s-%s-%s", "c-tenant", "c-ns", "c-fn");
-
+
factory = createKubernetesRuntimeFactory(null, 10, 1.0, 1.0);
factory.setCoreClient(coreApi);
factory.setAppsClient(appsApi);
ArgumentMatcher<String> hasTranslatedFunctionName = (String t) -> t.startsWith(expectedFunctionNamePrefix);
-
+
when(appsApi.deleteNamespacedStatefulSetCall(
argThat(hasTranslatedFunctionName),
anyString(), isNull(), isNull(), anyInt(), isNull(), anyString(), any(), isNull())).thenReturn(successfulCall);
@@ -1342,14 +1346,14 @@ public class KubernetesRuntimeTest {
V1PodList podList = mock(V1PodList.class);
when(podList.getItems()).thenReturn(Collections.emptyList());
-
+
String expectedLabels = String.format("tenant=%s,namespace=%s,name=%s", "c-tenant", "c-ns", "c-fn");
-
+
when(coreApi.listNamespacedPod(anyString(), isNull(), isNull(), isNull(), isNull(),
eq(expectedLabels), isNull(), isNull(), isNull(), isNull(), isNull())).thenReturn(podList);
- KubernetesRuntime kr = factory.createContainer(config, "/test/code", "code.yml", "/test/transforms", "transform.yml", Long.MIN_VALUE);
+ KubernetesRuntime kr = factory.createContainer(config, "/test/code", "code.yml", "/test/transforms", "transform.yml", Long.MIN_VALUE);
kr.deleteStatefulSet();
-
+
verify(coreApi).listNamespacedPod(anyString(), isNull(), isNull(), isNull(), isNull(),
eq(expectedLabels), isNull(), isNull(), isNull(), isNull(), isNull());
}
diff --git a/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/process/ProcessRuntimeTest.java b/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/process/ProcessRuntimeTest.java
index f63f24dc256..365704ea0b4 100644
--- a/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/process/ProcessRuntimeTest.java
+++ b/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/process/ProcessRuntimeTest.java
@@ -297,7 +297,7 @@ public class ProcessRuntimeTest {
String extraDepsEnv;
int portArg;
int metricsPortArg;
- int totalArgCount = 48;
+ int totalArgCount = 54;
if (webServiceUrl != null && config.isExposePulsarAdminClientEnabled()) {
totalArgCount += 3;
}
@@ -305,13 +305,13 @@ public class ProcessRuntimeTest {
assertEquals(args.size(), totalArgCount);
extraDepsEnv = " -Dpulsar.functions.extra.dependencies.dir=" + depsDir;
classpath = classpath + ":" + depsDir + "/*";
- portArg = 31;
- metricsPortArg = 33;
+ portArg = 37;
+ metricsPortArg = 39;
} else {
assertEquals(args.size(), totalArgCount-1);
extraDepsEnv = "";
- portArg = 30;
- metricsPortArg = 32;
+ portArg = 36;
+ metricsPortArg = 38;
}
if (webServiceUrl != null && config.isExposePulsarAdminClientEnabled()) {
portArg += 3;
@@ -328,7 +328,11 @@ public class ProcessRuntimeTest {
+ "-Dpulsar.function.log.dir=" + logDirectory + "/functions/" + FunctionCommon.getFullyQualifiedName(config.getFunctionDetails())
+ " -Dpulsar.function.log.file=" + config.getFunctionDetails().getName() + "-" + config.getInstanceId()
+ " -Dio.netty.tryReflectionSetAccessible=true"
- + " --add-opens java.base/sun.net=ALL-UNNAMED"
+ + " -Dorg.apache.pulsar.shade.io.netty.tryReflectionSetAccessible=true"
+ + " -Dio.grpc.netty.shaded.io.netty.tryReflectionSetAccessible=true"
+ + " --add-opens java.base/java.nio=ALL-UNNAMED"
+ + " --add-opens java.base/jdk.internal.misc=ALL-UNNAMED"
+ + " --add-opens java.base/java.util.zip=ALL-UNNAMED"
+ " org.apache.pulsar.functions.instance.JavaInstanceMain"
+ " --jar " + userJarFile
+ " --transform_function_jar " + userJarFile
(pulsar) 01/04: [fix][misc] Topic name from persistence name should decode local name (#22879)
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.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 0b3d8d52e21c081185394d724b997f4c33e3cb82
Author: 萧易客 <km...@live.com>
AuthorDate: Tue Jun 11 12:46:04 2024 +0800
[fix][misc] Topic name from persistence name should decode local name (#22879)
(cherry picked from commit c326d8e2203b6e9be37f4f2066fd7e90a9b9fb54)
---
.../src/main/java/org/apache/pulsar/common/naming/TopicName.java | 5 ++---
.../test/java/org/apache/pulsar/common/naming/TopicNameTest.java | 6 ++++++
2 files changed, 8 insertions(+), 3 deletions(-)
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java b/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java
index eebca0e0d72..e051e01495d 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java
@@ -358,17 +358,16 @@ public class TopicName implements ServiceUnitId {
String localName;
if (parts.size() == 4) {
tenant = parts.get(0);
- cluster = null;
namespacePortion = parts.get(1);
domain = parts.get(2);
- localName = parts.get(3);
+ localName = Codec.decode(parts.get(3));
return String.format("%s://%s/%s/%s", domain, tenant, namespacePortion, localName);
} else if (parts.size() == 5) {
tenant = parts.get(0);
cluster = parts.get(1);
namespacePortion = parts.get(2);
domain = parts.get(3);
- localName = parts.get(4);
+ localName = Codec.decode(parts.get(4));
return String.format("%s://%s/%s/%s/%s", domain, tenant, cluster, namespacePortion, localName);
} else {
throw new IllegalArgumentException("Invalid managedLedger name: " + mlName);
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/naming/TopicNameTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/naming/TopicNameTest.java
index 835045f9167..485bea3f1ad 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/naming/TopicNameTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/naming/TopicNameTest.java
@@ -267,6 +267,12 @@ public class TopicNameTest {
} catch (IllegalArgumentException e) {
// Exception is expected.
}
+
+ // case5: local name with special characters e.g. a:b:c
+ String topicName = "persistent://tenant/namespace/a:b:c";
+ String persistentNamingEncoding = "tenant/namespace/persistent/a%3Ab%3Ac";
+ assertEquals(TopicName.get(topicName).getPersistenceNamingEncoding(), persistentNamingEncoding);
+ assertEquals(TopicName.fromPersistenceNamingEncoding(persistentNamingEncoding), topicName);
}