You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by lc...@apache.org on 2021/11/18 21:57:47 UTC
[beam] branch revert-15857-java-wildcard created (now 1258b0d)
This is an automated email from the ASF dual-hosted git repository.
lcwik pushed a change to branch revert-15857-java-wildcard
in repository https://gitbox.apache.org/repos/asf/beam.git.
at 1258b0d Revert "Allow wildcards for java class lookup transform providers."
This branch includes the following new commits:
new 1258b0d Revert "Allow wildcards for java class lookup transform providers."
The 1 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.
[beam] 01/01: Revert "Allow wildcards for java class lookup transform providers."
Posted by lc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lcwik pushed a commit to branch revert-15857-java-wildcard
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 1258b0d177e9cd74360af6632aa38b4cb18ab6c4
Author: Lukasz Cwik <lc...@google.com>
AuthorDate: Thu Nov 18 13:56:48 2021 -0800
Revert "Allow wildcards for java class lookup transform providers."
---
sdks/java/expansion-service/build.gradle | 9 +--
.../expansion/service/ExpansionServiceOptions.java | 10 ++-
.../service/JavaClassLookupTransformProvider.java | 83 ++++++----------------
.../JavaClassLookupTransformProviderTest.java | 42 -----------
sdks/python/apache_beam/transforms/external.py | 56 +++------------
.../python/apache_beam/transforms/external_test.py | 24 -------
sdks/python/apache_beam/utils/processes_test.py | 8 ---
sdks/python/apache_beam/utils/subprocess_server.py | 47 +-----------
.../apache_beam/utils/subprocess_server_test.py | 51 -------------
9 files changed, 37 insertions(+), 293 deletions(-)
diff --git a/sdks/java/expansion-service/build.gradle b/sdks/java/expansion-service/build.gradle
index 67336b4..2a0ffd0 100644
--- a/sdks/java/expansion-service/build.gradle
+++ b/sdks/java/expansion-service/build.gradle
@@ -16,14 +16,9 @@
* limitations under the License.
*/
-apply plugin: 'org.apache.beam.module'
-apply plugin: 'application'
+plugins { id 'org.apache.beam.module' }
applyJavaNature(
- automaticModuleName: 'org.apache.beam.sdk.expansion.service',
- validateShadowJar: false,
- shadowClosure: {},
-)
-mainClassName = "org.apache.beam.sdk.expansion.service.ExpansionService"
+ automaticModuleName: 'org.apache.beam.sdk.expansion.service')
description = "Apache Beam :: SDKs :: Java :: Expansion Service"
ext.summary = """Contains code that can be used to run an expansion service."""
diff --git a/sdks/java/expansion-service/src/main/java/org/apache/beam/sdk/expansion/service/ExpansionServiceOptions.java b/sdks/java/expansion-service/src/main/java/org/apache/beam/sdk/expansion/service/ExpansionServiceOptions.java
index db66bd9..79e870c 100644
--- a/sdks/java/expansion-service/src/main/java/org/apache/beam/sdk/expansion/service/ExpansionServiceOptions.java
+++ b/sdks/java/expansion-service/src/main/java/org/apache/beam/sdk/expansion/service/ExpansionServiceOptions.java
@@ -21,6 +21,7 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
import java.io.File;
import java.io.IOException;
+import java.util.ArrayList;
import org.apache.beam.sdk.expansion.service.JavaClassLookupTransformProvider.AllowList;
import org.apache.beam.sdk.options.Default;
import org.apache.beam.sdk.options.DefaultValueFactory;
@@ -36,8 +37,7 @@ public interface ExpansionServiceOptions extends PipelineOptions {
void setJavaClassLookupAllowlist(AllowList file);
- @Description(
- "Allow list file for Java class based transform expansion, or '*' to allow anything.")
+ @Description("Allow list file for Java class based transform expansion")
String getJavaClassLookupAllowlistFile();
void setJavaClassLookupAllowlistFile(String file);
@@ -53,9 +53,6 @@ public interface ExpansionServiceOptions extends PipelineOptions {
String allowListFile =
options.as(ExpansionServiceOptions.class).getJavaClassLookupAllowlistFile();
if (allowListFile != null) {
- if (allowListFile.equals("*")) {
- return AllowList.everything();
- }
ObjectMapper mapper = new ObjectMapper(new YAMLFactory());
File allowListFileObj = new File(allowListFile);
if (!allowListFileObj.exists()) {
@@ -71,7 +68,8 @@ public interface ExpansionServiceOptions extends PipelineOptions {
}
// By default produces an empty allow-list.
- return AllowList.nothing();
+ return new AutoValue_JavaClassLookupTransformProvider_AllowList(
+ JavaClassLookupTransformProvider.ALLOW_LIST_VERSION, new ArrayList<>());
}
}
}
diff --git a/sdks/java/expansion-service/src/main/java/org/apache/beam/sdk/expansion/service/JavaClassLookupTransformProvider.java b/sdks/java/expansion-service/src/main/java/org/apache/beam/sdk/expansion/service/JavaClassLookupTransformProvider.java
index 2a35b2c..04d0b77 100644
--- a/sdks/java/expansion-service/src/main/java/org/apache/beam/sdk/expansion/service/JavaClassLookupTransformProvider.java
+++ b/sdks/java/expansion-service/src/main/java/org/apache/beam/sdk/expansion/service/JavaClassLookupTransformProvider.java
@@ -34,7 +34,6 @@ import java.lang.reflect.Type;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
-import java.util.Collections;
import java.util.List;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
@@ -101,7 +100,24 @@ class JavaClassLookupTransformProvider<InputT extends PInput, OutputT extends PO
String className = payload.getClassName();
try {
- AllowedClass allowlistClass = allowList.getAllowedClass(className);
+ AllowedClass allowlistClass = null;
+ if (this.allowList != null) {
+ for (AllowedClass cls : this.allowList.getAllowedClasses()) {
+ if (cls.getClassName().equals(className)) {
+ if (allowlistClass != null) {
+ throw new IllegalArgumentException(
+ "Found two matching allowlist classes " + allowlistClass + " and " + cls);
+ }
+ allowlistClass = cls;
+ }
+ }
+ }
+ if (allowlistClass == null) {
+ throw new UnsupportedOperationException(
+ "The provided allow list does not enable expanding a transform class by the name "
+ + className
+ + ".");
+ }
Class<PTransform<InputT, OutputT>> transformClass =
(Class<PTransform<InputT, OutputT>>)
ReflectHelpers.findClassLoader().loadClass(className);
@@ -174,7 +190,7 @@ class JavaClassLookupTransformProvider<InputT extends PInput, OutputT extends PO
for (Annotation annotation : method.getAnnotations()) {
if (annotation instanceof MultiLanguageBuilderMethod) {
if (nameFromPayload.equals(((MultiLanguageBuilderMethod) annotation).name())) {
- if (allowListClass.isAllowedBuilderMethod(nameFromPayload)) {
+ if (allowListClass.getAllowedBuilderMethods().contains(nameFromPayload)) {
return true;
} else {
throw new RuntimeException(
@@ -197,7 +213,7 @@ class JavaClassLookupTransformProvider<InputT extends PInput, OutputT extends PO
consideredMethodName.substring(4, 5).toLowerCase() + consideredMethodName.substring(5);
match = consideredMethodName.equals(nameFromPayload);
}
- if (match && !allowListClass.isAllowedBuilderMethod(consideredMethodName)) {
+ if (match && !allowListClass.getAllowedBuilderMethods().contains(consideredMethodName)) {
throw new RuntimeException(
"Builder method name " + consideredMethodName + " has to be explicitly allowed");
}
@@ -423,7 +439,7 @@ class JavaClassLookupTransformProvider<InputT extends PInput, OutputT extends PO
for (Annotation annotation : method.getAnnotations()) {
if (annotation instanceof MultiLanguageConstructorMethod) {
if (nameFromPayload.equals(((MultiLanguageConstructorMethod) annotation).name())) {
- if (allowListClass.isAllowedConstructorMethod(nameFromPayload)) {
+ if (allowListClass.getAllowedConstructorMethods().contains(nameFromPayload)) {
return true;
} else {
throw new RuntimeException(
@@ -433,7 +449,7 @@ class JavaClassLookupTransformProvider<InputT extends PInput, OutputT extends PO
}
}
if (method.getName().equals(nameFromPayload)) {
- if (allowListClass.isAllowedConstructorMethod(nameFromPayload)) {
+ if (allowListClass.getAllowedConstructorMethods().contains(nameFromPayload)) {
return true;
} else {
throw new RuntimeException(
@@ -473,41 +489,10 @@ class JavaClassLookupTransformProvider<InputT extends PInput, OutputT extends PO
@AutoValue
public abstract static class AllowList {
- public static AllowList nothing() {
- return create(ALLOW_LIST_VERSION, Collections.emptyList());
- }
-
- public static AllowList everything() {
- return create(
- ALLOW_LIST_VERSION,
- Collections.singletonList(
- AllowedClass.create("*", AllowedClass.WILDCARD, AllowedClass.WILDCARD)));
- }
-
public abstract String getVersion();
public abstract List<AllowedClass> getAllowedClasses();
- public AllowedClass getAllowedClass(String className) {
- AllowedClass allowlistClass = null;
- for (AllowedClass cls : getAllowedClasses()) {
- if (cls.isAllowedClass(className)) {
- if (allowlistClass != null) {
- throw new IllegalArgumentException(
- "Found two matching allowlist classes " + allowlistClass + " and " + cls);
- }
- allowlistClass = cls;
- }
- }
- if (allowlistClass == null) {
- throw new UnsupportedOperationException(
- "The provided allow list does not enable expanding a transform class by the name "
- + className
- + ".");
- }
- return allowlistClass;
- }
-
@JsonCreator
static AllowList create(
@JsonProperty("version") String version,
@@ -523,32 +508,12 @@ class JavaClassLookupTransformProvider<InputT extends PInput, OutputT extends PO
@AutoValue
public abstract static class AllowedClass {
- public static final List<String> WILDCARD = Collections.singletonList("*");
-
public abstract String getClassName();
public abstract List<String> getAllowedBuilderMethods();
public abstract List<String> getAllowedConstructorMethods();
- public boolean isAllowedClass(String className) {
- String pattern = getClassName();
- return pattern.equals(className)
- || pattern.equals("*")
- || (pattern.endsWith(".*")
- && className.startsWith(pattern.substring(0, pattern.length() - 2)));
- }
-
- public boolean isAllowedBuilderMethod(String methodName) {
- return getAllowedBuilderMethods().contains(methodName)
- || getAllowedBuilderMethods().equals(WILDCARD);
- }
-
- public boolean isAllowedConstructorMethod(String methodName) {
- return getAllowedConstructorMethods().contains(methodName)
- || getAllowedConstructorMethods().equals(WILDCARD);
- }
-
@JsonCreator
static AllowedClass create(
@JsonProperty("className") String className,
@@ -562,10 +527,6 @@ class JavaClassLookupTransformProvider<InputT extends PInput, OutputT extends PO
if (allowedConstructorMethods == null) {
allowedConstructorMethods = new ArrayList<>();
}
- if (allowedBuilderMethods.equals(WILDCARD) && !className.equals("*")) {
- // If we allow getClass().forName(), we allow essentially anything.
- throw new IllegalArgumentException("Wildcard builder not allowed for non-wildcard class.");
- }
return new AutoValue_JavaClassLookupTransformProvider_AllowedClass(
className, allowedBuilderMethods, allowedConstructorMethods);
}
diff --git a/sdks/java/expansion-service/src/test/java/org/apache/beam/sdk/expansion/service/JavaClassLookupTransformProviderTest.java b/sdks/java/expansion-service/src/test/java/org/apache/beam/sdk/expansion/service/JavaClassLookupTransformProviderTest.java
index bb78adc..08e9ed0 100644
--- a/sdks/java/expansion-service/src/test/java/org/apache/beam/sdk/expansion/service/JavaClassLookupTransformProviderTest.java
+++ b/sdks/java/expansion-service/src/test/java/org/apache/beam/sdk/expansion/service/JavaClassLookupTransformProviderTest.java
@@ -18,15 +18,11 @@
package org.apache.beam.sdk.expansion.service;
import static org.apache.beam.runners.core.construction.BeamUrns.getUrn;
-import static org.apache.beam.sdk.expansion.service.JavaClassLookupTransformProvider.ALLOW_LIST_VERSION;
-import static org.apache.beam.sdk.expansion.service.JavaClassLookupTransformProvider.AllowList;
-import static org.apache.beam.sdk.expansion.service.JavaClassLookupTransformProvider.AllowedClass;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.anyOf;
import static org.hamcrest.Matchers.containsString;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;
@@ -36,7 +32,6 @@ import java.io.IOException;
import java.io.Serializable;
import java.net.URL;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.List;
import java.util.Map;
import org.apache.beam.model.expansion.v1.ExpansionApi;
@@ -1155,41 +1150,4 @@ public class JavaClassLookupTransformProviderTest {
return outputStream.toByteString();
}
-
- @Test
- public void testNothingAllowList() {
- AllowList nothing = AllowList.nothing();
- RuntimeException thrown =
- assertThrows(
- RuntimeException.class,
- () -> nothing.getAllowedClass("org.apache.beam.sdk.transforms.KvSwap"));
- assertTrue(thrown.getMessage(), thrown.getMessage().contains("allow list does not enable"));
- assertTrue(
- thrown.getMessage(), thrown.getMessage().contains("org.apache.beam.sdk.transforms.KvSwap"));
- }
-
- @Test
- public void testEverythingAllowList() {
- AllowList everything = AllowList.everything();
- AllowedClass allowedClass = everything.getAllowedClass("org.apache.beam.sdk.transforms.KvSwap");
- assertTrue(allowedClass.isAllowedBuilderMethod("builder"));
- assertTrue(allowedClass.isAllowedConstructorMethod("constructor"));
- }
-
- @Test
- public void testPackageAllowList() {
- AllowList allowList =
- AllowList.create(
- ALLOW_LIST_VERSION,
- Collections.singletonList(
- AllowedClass.create(
- "good.package.*",
- Collections.singletonList("goodBuilder"),
- AllowedClass.WILDCARD)));
- assertThrows(RuntimeException.class, () -> allowList.getAllowedClass("bad.package.Transform"));
- AllowedClass allowedClass = allowList.getAllowedClass("good.package.Transform");
- assertTrue(allowedClass.isAllowedBuilderMethod("goodBuilder"));
- assertFalse(allowedClass.isAllowedBuilderMethod("badBuilder"));
- assertTrue(allowedClass.isAllowedConstructorMethod("anyConstructor"));
- }
}
diff --git a/sdks/python/apache_beam/transforms/external.py b/sdks/python/apache_beam/transforms/external.py
index a64cfae..008b076 100644
--- a/sdks/python/apache_beam/transforms/external.py
+++ b/sdks/python/apache_beam/transforms/external.py
@@ -24,7 +24,6 @@ No backward compatibility guarantees. Everything in this module is experimental.
import contextlib
import copy
import functools
-import glob
import threading
from collections import OrderedDict
from typing import Dict
@@ -284,30 +283,12 @@ class JavaClassLookupPayloadBuilder(PayloadBuilder):
class JavaExternalTransform(ptransform.PTransform):
"""A proxy for Java-implemented external transforms.
- One builds these transforms just as one would in Java, e.g.::
-
- transform = JavaExternalTransform('fully.qualified.ClassName'
- )(contructorArg, ... ).builderMethod(...)
-
- or::
-
- JavaExternalTransform('fully.qualified.ClassName').staticConstructor(
- ...).builderMethod1(...).builderMethod2(...)
-
- :param class_name: fully qualified name of the java class
- :param expansion_service: (Optional) an expansion service to use. If none is
- provided, a default expansion service will be started.
- :param classpath: (Optional) A list paths to additional jars to place on the
- expansion service classpath.
+ One builds these transforms just as one would in Java.
"""
- def __init__(self, class_name, expansion_service=None, classpath=None):
- if expansion_service and classpath:
- raise ValueError(
- f'Only one of expansion_service ({expansion_service}) '
- f'or classpath ({classpath}) may be provided.')
+ def __init__(self, class_name, expansion_service=None):
self._payload_builder = JavaClassLookupPayloadBuilder(class_name)
- self._classpath = classpath
self._expansion_service = expansion_service
+
# Beam explicitly looks for following attributes. Hence adding
# 'None' values here to prevent '__getattr__' from being called.
self.inputs = None
@@ -321,11 +302,7 @@ class JavaExternalTransform(ptransform.PTransform):
# Don't try to emulate special methods.
if name.startswith('__') and name.endswith('__'):
return super().__getattr__(name)
- else:
- return self[name]
- def __getitem__(self, name):
- # Use directly for keywords or attribute conflicts.
def construct(*args, **kwargs):
if self._payload_builder._has_constructor():
builder_method = self._payload_builder.add_builder_method
@@ -337,11 +314,6 @@ class JavaExternalTransform(ptransform.PTransform):
return construct
def expand(self, pcolls):
- if self._expansion_service is None:
- self._expansion_service = BeamJarExpansionService(
- ':sdks:java:expansion-service:shadowJar',
- extra_args=['{{PORT}}', '--javaClassLookupAllowlistFile=*'],
- classpath=self._classpath)
return pcolls | ExternalTransform(
common_urns.java_class_lookup.urn,
self._payload_builder,
@@ -683,29 +655,22 @@ class JavaJarExpansionService(object):
argument which will spawn a subprocess using this jar to expand the
transform.
"""
- def __init__(self, path_to_jar, extra_args=None, classpath=None):
+ def __init__(self, path_to_jar, extra_args=None):
self._path_to_jar = path_to_jar
self._extra_args = extra_args
- self._classpath = classpath
self._service_count = 0
- def _default_args(self):
- to_stage = ','.join([self._path_to_jar] + sum(
- (glob.glob(path) or [path] for path in self._classpath or []), []))
- return ['{{PORT}}', f'--filesToStage={to_stage}']
-
def __enter__(self):
if self._service_count == 0:
self._path_to_jar = subprocess_server.JavaJarServer.local_jar(
self._path_to_jar)
if self._extra_args is None:
- self._extra_args = self._default_args()
+ self._extra_args = ['{{PORT}}', f'--filesToStage={self._path_to_jar}']
# Consider memoizing these servers (with some timeout).
self._service_provider = subprocess_server.JavaJarServer(
ExpansionAndArtifactRetrievalStub,
self._path_to_jar,
- self._extra_args,
- classpath=self._classpath)
+ self._extra_args)
self._service = self._service_provider.__enter__()
self._service_count += 1
return self._service
@@ -723,15 +688,10 @@ class BeamJarExpansionService(JavaJarExpansionService):
if it exists, otherwise attempts to download and cache the released artifact
corresponding to this version of Beam from the apache maven repository.
"""
- def __init__(
- self,
- gradle_target,
- extra_args=None,
- gradle_appendix=None,
- classpath=None):
+ def __init__(self, gradle_target, extra_args=None, gradle_appendix=None):
path_to_jar = subprocess_server.JavaJarServer.path_to_beam_jar(
gradle_target, gradle_appendix)
- super().__init__(path_to_jar, extra_args, classpath=classpath)
+ super().__init__(path_to_jar, extra_args)
def memoize(func):
diff --git a/sdks/python/apache_beam/transforms/external_test.py b/sdks/python/apache_beam/transforms/external_test.py
index c2e3932..83e7246 100644
--- a/sdks/python/apache_beam/transforms/external_test.py
+++ b/sdks/python/apache_beam/transforms/external_test.py
@@ -21,8 +21,6 @@
import dataclasses
import logging
-import os
-import tempfile
import typing
import unittest
@@ -42,7 +40,6 @@ from apache_beam.transforms.external import AnnotationBasedPayloadBuilder
from apache_beam.transforms.external import ImplicitSchemaPayloadBuilder
from apache_beam.transforms.external import JavaClassLookupPayloadBuilder
from apache_beam.transforms.external import JavaExternalTransform
-from apache_beam.transforms.external import JavaJarExpansionService
from apache_beam.transforms.external import NamedTupleBasedPayloadBuilder
from apache_beam.typehints import typehints
from apache_beam.typehints.native_type_compatibility import convert_to_beam_type
@@ -550,27 +547,6 @@ class JavaClassLookupPayloadBuilderTest(unittest.TestCase):
self._verify_row(build_method.schema, build_method.payload, {})
-class JavaJarExpansionServiceTest(unittest.TestCase):
- def test_classpath(self):
- with tempfile.TemporaryDirectory() as temp_dir:
- try:
- # Avoid having to prefix everything in our test strings.
- oldwd = os.getcwd()
- os.chdir(temp_dir)
- # Touch some files for globing.
- with open('a1.jar', 'w') as _:
- pass
-
- service = JavaJarExpansionService(
- 'main.jar', classpath=['a*.jar', 'b.jar'])
- self.assertEqual(
- service._default_args(),
- ['{{PORT}}', '--filesToStage=main.jar,a1.jar,b.jar'])
-
- finally:
- os.chdir(oldwd)
-
-
if __name__ == '__main__':
logging.getLogger().setLevel(logging.INFO)
unittest.main()
diff --git a/sdks/python/apache_beam/utils/processes_test.py b/sdks/python/apache_beam/utils/processes_test.py
index 3091ebc..c2868eb 100644
--- a/sdks/python/apache_beam/utils/processes_test.py
+++ b/sdks/python/apache_beam/utils/processes_test.py
@@ -143,10 +143,6 @@ class TestErrorHandlingCheckOutput(unittest.TestCase):
'apache_beam.utils.processes.subprocess.check_output')
cls.mock_get = cls.mock_get_patcher.start()
- @classmethod
- def tearDownClass(cls):
- cls.mock_get_patcher.stop()
-
def test_oserror_check_output_message(self):
self.mock_get.side_effect = OSError()
cmd = ["lls"]
@@ -184,10 +180,6 @@ class TestErrorHandlingCall(unittest.TestCase):
'apache_beam.utils.processes.subprocess.call')
cls.mock_get = cls.mock_get_patcher.start()
- @classmethod
- def tearDownClass(cls):
- cls.mock_get_patcher.stop()
-
def test_oserror_check_output_message(self):
self.mock_get.side_effect = OSError()
cmd = ["lls"]
diff --git a/sdks/python/apache_beam/utils/subprocess_server.py b/sdks/python/apache_beam/utils/subprocess_server.py
index 7f6c24d..7035cad 100644
--- a/sdks/python/apache_beam/utils/subprocess_server.py
+++ b/sdks/python/apache_beam/utils/subprocess_server.py
@@ -18,8 +18,6 @@
# pytype: skip-file
import contextlib
-import glob
-import hashlib
import logging
import os
import re
@@ -30,7 +28,6 @@ import subprocess
import tempfile
import threading
import time
-import zipfile
from urllib.error import URLError
from urllib.request import urlopen
@@ -161,11 +158,7 @@ class JavaJarServer(SubprocessServer):
'local', (threading.local, ),
dict(__init__=lambda self: setattr(self, 'replacements', {})))()
- def __init__(self, stub_class, path_to_jar, java_arguments, classpath=None):
- if classpath:
- # java -jar ignores the classpath, so we make a new jar that embeds
- # the requested classpath.
- path_to_jar = self.make_classpath_jar(path_to_jar, classpath)
+ def __init__(self, stub_class, path_to_jar, java_arguments):
super().__init__(
stub_class, ['java', '-jar', path_to_jar] + list(java_arguments))
self._existing_service = path_to_jar if _is_service_endpoint(
@@ -290,44 +283,6 @@ class JavaJarServer(SubprocessServer):
finally:
cls._BEAM_SERVICES.replacements = old
- @classmethod
- def make_classpath_jar(cls, main_jar, extra_jars, cache_dir=None):
- if cache_dir is None:
- cache_dir = cls.JAR_CACHE
- composite_jar_dir = os.path.join(cache_dir, 'composite-jars')
- os.makedirs(composite_jar_dir, exist_ok=True)
- classpath = []
- # Class-Path references from a jar must be relative, so we create
- # a relatively-addressable subdirectory with symlinks to all the
- # required jars.
- for pattern in [main_jar] + list(extra_jars):
- for path in glob.glob(pattern) or [pattern]:
- path = os.path.abspath(path)
- rel_path = hashlib.sha256(
- path.encode('utf-8')).hexdigest() + os.path.splitext(path)[1]
- classpath.append(rel_path)
- if not os.path.lexists(os.path.join(composite_jar_dir, rel_path)):
- os.symlink(path, os.path.join(composite_jar_dir, rel_path))
- # Now create a single jar that simply references the rest and has the same
- # main class as main_jar.
- composite_jar = os.path.join(
- composite_jar_dir,
- hashlib.sha256(' '.join(sorted(classpath)).encode('ascii')).hexdigest()
- + '.jar')
- if not os.path.exists(composite_jar):
- with zipfile.ZipFile(main_jar) as main:
- with main.open('META-INF/MANIFEST.MF') as manifest:
- main_class = next(
- filter(lambda line: line.startswith(b'Main-Class: '), manifest))
- with zipfile.ZipFile(composite_jar + '.tmp', 'w') as composite:
- with composite.open('META-INF/MANIFEST.MF', 'w') as manifest:
- manifest.write(b'Manifest-Version: 1.0\n')
- manifest.write(main_class)
- manifest.write(
- b'Class-Path: ' + ' '.join(classpath).encode('ascii') + b'\n')
- os.rename(composite_jar + '.tmp', composite_jar)
- return composite_jar
-
def _is_service_endpoint(path):
return re.match(r'^[a-zA-Z0-9.-]+:\d+$', path)
diff --git a/sdks/python/apache_beam/utils/subprocess_server_test.py b/sdks/python/apache_beam/utils/subprocess_server_test.py
index e0d0892..025eed4 100644
--- a/sdks/python/apache_beam/utils/subprocess_server_test.py
+++ b/sdks/python/apache_beam/utils/subprocess_server_test.py
@@ -19,12 +19,9 @@
# pytype: skip-file
-import glob
import os
import re
-import shutil
import socketserver
-import subprocess
import tempfile
import threading
import unittest
@@ -117,54 +114,6 @@ class JavaJarServerTest(unittest.TestCase):
with open(os.path.join(temp_dir, 'file.jar')) as fin:
self.assertEqual(fin.read(), 'data')
- @unittest.skipUnless(shutil.which('javac'), 'missing java jdk')
- def test_classpath_jar(self):
- with tempfile.TemporaryDirectory() as temp_dir:
- try:
- # Avoid having to prefix everything in our test strings.
- oldwd = os.getcwd()
- os.chdir(temp_dir)
-
- with open('Main.java', 'w') as fout:
- fout.write(
- """
-public class Main {
- public static void main(String[] args) { Other.greet(); }
-}
- """)
-
- with open('Other.java', 'w') as fout:
- fout.write(
- """
-public class Other {
- public static void greet() { System.out.println("You got me!"); }
-}
- """)
-
- os.mkdir('jars')
- # Using split just for readability/copyability.
- subprocess.check_call('javac Main.java Other.java'.split())
- subprocess.check_call('jar cfe jars/Main.jar Main Main.class'.split())
- subprocess.check_call('jar cf jars/Other.jar Other.class'.split())
- # Make sure the java and class files don't get picked up.
- for path in glob.glob('*.*'):
- os.unlink(path)
-
- # These should fail.
- self.assertNotEqual(
- subprocess.call('java -jar jars/Main.jar'.split()), 0)
- self.assertNotEqual(
- subprocess.call('java -jar jars/Other.jar'.split()), 0)
-
- os.mkdir('beam_temp')
- composite_jar = subprocess_server.JavaJarServer.make_classpath_jar(
- 'jars/Main.jar', ['jars/Other.jar'], cache_dir='beam_temp')
- # This, however, should work.
- subprocess.check_call(f'java -jar {composite_jar}'.split())
-
- finally:
- os.chdir(oldwd)
-
if __name__ == '__main__':
unittest.main()