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:08 UTC
(pulsar) 03/04: [fix][fn] Enable optimized Netty direct byte buffer support for Pulsar Function runtimes (#22910)
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