streamnative · freeznet · Mar 21, 2024 · Mar 26, 2024
@@ -342,11 +342,11 @@ func MakeJavaFunctionCommand(downloadPath, packageFile, name, clusterName, gener
 	javaOpts []string, hasPulsarctl, hasWget, authProvided, tlsProvided bool, secretMaps map[string]v1alpha1.SecretRef,
 	state *v1alpha1.Stateful,
 	tlsConfig TLSConfig, authConfig *v1alpha1.AuthConfig,
-	maxPendingAsyncRequests *int32, logConfigFileName string) []string {
+	maxPendingAsyncRequests *int32, logConfigFileName string, java9orAbove bool) []string {
 	processCommand := setShardIDEnvironmentVariableCommand() + " && " + generateLogConfigCommand +
 		strings.Join(getProcessJavaRuntimeArgs(name, packageFile, clusterName, logLevel, details,
 			extraDependenciesDir, uid, javaOpts, authProvided, tlsProvided, secretMaps, state, tlsConfig,
-			authConfig, maxPendingAsyncRequests, logConfigFileName), " ")
+			authConfig, maxPendingAsyncRequests, logConfigFileName, java9orAbove), " ")
 	if downloadPath != "" && !utils.EnableInitContainers {
 		// prepend download command if the downPath is provided
 		downloadCommand := strings.Join(getDownloadCommand(downloadPath, packageFile, hasPulsarctl, hasWget,
@@ -1092,7 +1092,7 @@ func getProcessJavaRuntimeArgs(name, packageName, clusterName, logLevel, details
 	javaOpts []string, authProvided, tlsProvided bool, secretMaps map[string]v1alpha1.SecretRef,
 	state *v1alpha1.Stateful,
 	tlsConfig TLSConfig, authConfig *v1alpha1.AuthConfig,
-	maxPendingAsyncRequests *int32, logConfigFileName string) []string {
+	maxPendingAsyncRequests *int32, logConfigFileName string, java9orAbove bool) []string {
 	classPath := "/pulsar/instances/java-instance.jar:/pulsar/lib/*"
 	javaLogConfigPath := logConfigFileName
 	if javaLogConfigPath == "" {
@@ -1121,17 +1121,23 @@ func getProcessJavaRuntimeArgs(name, packageName, clusterName, logLevel, details
 		"-Dpulsar.function.log.dir=logs/functions",
 		"-Dpulsar.function.log.file=" + fmt.Sprintf("%s-${%s}", name, EnvShardID),
 		setLogLevel,
+		"-Dio.netty.tryReflectionSetAccessible=true",
 		"-XX:InitialRAMPercentage=20",
 		"-XX:MaxRAMPercentage=40",
 		"-XX:+UseG1GC",
 		"-XX:+HeapDumpOnOutOfMemoryError",
 		"-XX:HeapDumpPath=/pulsar/tmp/heapdump-%p.hprof",
 		"-Xlog:gc*:file=/pulsar/logs/gc.log:time,level,tags:filecount=5,filesize=10M",
-		strings.Join(javaOpts, " "),
-		"org.apache.pulsar.functions.instance.JavaInstanceMain",
-		"--jar",
-		packageName,
 	}
+	args = append(args, javaOpts...)
+	if java9orAbove {
+		// Needed for netty.DnsResolverUtil on JDK9+
+		args = append(args, "--add-opens java.base/sun.net=ALL-UNNAMED")
+	}
+	args = append(args, "org.apache.pulsar.functions.instance.JavaInstanceMain",
+		"--jar",
+		packageName)
+
 	sharedArgs := getSharedArgs(details, clusterName, uid, authProvided, tlsProvided, tlsConfig, authConfig)
 	args = append(args, sharedArgs...)
 	if len(secretMaps) > 0 {
@@ -2276,3 +2282,38 @@ func makeFilebeatContainer(volumeMounts []corev1.VolumeMount, envVar []corev1.En
 		},
 	}
 }
+
+// extract docker image tag from image name
+func extractImageTag(image string) string {
+	if image == "" {
+		return "latest"
+	}
+	s := strings.Split(image, ":")
+	if len(s) == 2 {
+		return s[1]
+	}
+	return "latest"
+}
+
+// check if the image tag is for jre 9+
+func isJava9orAbove(tag string) bool {
+	if tag == "latest" {
+		return true
+	}
+	if strings.Contains(tag, ".") {
+		s := strings.Split(tag, ".")
+		if len(s) > 1 {
+			major, err := strconv.Atoi(s[0])
+			if err != nil {
+				return true
+			}
+			minor, err := strconv.Atoi(s[1])
+			if err != nil {
+				return true
+			}
+
+			return (major == 2 && minor >= 9) || major >= 3
+		}
+	}
+	return true
+}
@@ -212,6 +212,8 @@ func makeFunctionCommand(function *v1alpha1.Function) []string {
 		hasPulsarctl = true
 		hasWget = true
 	}
+	imageVersion := extractImageTag(function.Spec.Image)
+	java9orAbove := isJava9orAbove(imageVersion)
 	if spec.Java != nil {
 		if spec.Java.Jar != "" {
 			return MakeJavaFunctionCommand(spec.Java.JarLocation, spec.Java.Jar,
@@ -225,7 +227,8 @@ func makeFunctionCommand(function *v1alpha1.Function) []string {
 				spec.Pulsar.AuthSecret != "", spec.Pulsar.TLSSecret != "",
 				spec.SecretsMap, spec.StateConfig, spec.Pulsar.TLSConfig,
 				spec.Pulsar.AuthConfig, spec.MaxPendingAsyncRequests,
-				generateJavaLogConfigFileName(function.Spec.Java))
+				generateJavaLogConfigFileName(function.Spec.Java),
+				java9orAbove)
 		}
 	} else if spec.Python != nil {
 		if spec.Python.Py != "" {

@@ -207,6 +207,8 @@ func MakeSinkCommand(sink *v1alpha1.Sink) []string {
 		hasPulsarctl = true
 		hasWget = true
 	}
+	imageVersion := extractImageTag(sink.Spec.Image)
+	java9orAbove := isJava9orAbove(imageVersion)
 	return MakeJavaFunctionCommand(spec.Java.JarLocation, spec.Java.Jar,
 		spec.Name, spec.ClusterName,
 		generateJavaLogConfigCommand(spec.Java, spec.LogTopicAgent),
@@ -215,7 +217,7 @@ func MakeSinkCommand(sink *v1alpha1.Sink) []string {
 		spec.Java.ExtraDependenciesDir, string(sink.UID),
 		spec.Java.JavaOpts, hasPulsarctl, hasWget, spec.Pulsar.AuthSecret != "", spec.Pulsar.TLSSecret != "",
 		spec.SecretsMap, spec.StateConfig, spec.Pulsar.TLSConfig, spec.Pulsar.AuthConfig, nil,
-		generateJavaLogConfigFileName(spec.Java))
+		generateJavaLogConfigFileName(spec.Java), java9orAbove)
 }

 func generateSinkDetailsInJSON(sink *v1alpha1.Sink) string {

@@ -154,6 +154,8 @@ func makeSourceCommand(source *v1alpha1.Source) []string {
 		hasPulsarctl = true
 		hasWget = true
 	}
+	imageVersion := extractImageTag(source.Spec.Image)
+	java9orAbove := isJava9orAbove(imageVersion)
 	return MakeJavaFunctionCommand(spec.Java.JarLocation, spec.Java.Jar,
 		spec.Name, spec.ClusterName,
 		generateJavaLogConfigCommand(spec.Java, spec.LogTopicAgent),
@@ -162,7 +164,7 @@ func makeSourceCommand(source *v1alpha1.Source) []string {
 		spec.Java.ExtraDependenciesDir, string(source.UID),
 		spec.Java.JavaOpts, hasPulsarctl, hasWget, spec.Pulsar.AuthSecret != "", spec.Pulsar.TLSSecret != "",
 		spec.SecretsMap, spec.StateConfig, spec.Pulsar.TLSConfig, spec.Pulsar.AuthConfig, nil,
-		generateJavaLogConfigFileName(spec.Java))
+		generateJavaLogConfigFileName(spec.Java), java9orAbove)
 }

 func generateSourceDetailsInJSON(source *v1alpha1.Source) string {

@@ -10,7 +10,7 @@ require (
 	github.com/prometheus/client_golang v1.12.2
 	github.com/streamnative/pulsarctl v0.4.3-0.20220702165443-e4c26e2c39cf
 	github.com/stretchr/testify v1.7.0
-	google.golang.org/protobuf v1.28.0
+	google.golang.org/protobuf v1.33.0
 	gotest.tools v2.2.0+incompatible
 	k8s.io/api v0.25.0
 	k8s.io/apimachinery v0.25.0

@@ -840,8 +840,8 @@ google.golang.org/protobuf v1.25.0/go.mod h1:9JNX74DMeImyA3h4bdi1ymwjUzf21/xIlba
 google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
 google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
 google.golang.org/protobuf v1.27.1/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
-google.golang.org/protobuf v1.28.0 h1:w43yiav+6bVFTBQFZX0r7ipe9JQ1QsbMgHwbBziscLw=
-google.golang.org/protobuf v1.28.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
+google.golang.org/protobuf v1.33.0 h1:uNO2rsAINq/JlFpSdYEKIZ0uKD/R9cpdv0T+yoGwGmI=
+google.golang.org/protobuf v1.33.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=
 gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=