Merge pull request #988: Fix direct classloader

Author: je-ik

beam/tools/src/main/java/cz/o2/proxima/beam/tools/groovy/RemoteConsumer.java

@@ -103,7 +103,7 @@ public StreamObserver<Item> collect(StreamObserver<Response> responseObserver) {
       unterminatedCalls.add(terminateFuture);
       // send initial status to client
       responseObserver.onNext(Response.newBuilder().setStatus(CONTINUE).build());
-      return new StreamObserver<Item>() {
+      return new StreamObserver<>() {
 
         List<Item> received = new ArrayList<>();
 

core/src/main/java/cz/o2/proxima/core/repository/JPMSMain.java

@@ -64,6 +64,7 @@ public static void main(String[] args)
         new ChildLayerFirstClassLoader(finder, thisClass.getClassLoader());
     createModuleLayer(loaderCache, parentLayer, finder, parentLoader);
     log.info("Added module layer from path {} with modules {}", path, finder.findAll());
+    Repository.setJpmsClassLoader(parentLoader);
     Thread.currentThread().setContextClassLoader(parentLoader);
     Class<?> loadedMainClass = parentLoader.loadClass(mainClass);
     Method main = loadedMainClass.getDeclaredMethod("main", String[].class);

core/src/main/java/cz/o2/proxima/core/repository/Repository.java

@@ -18,6 +18,7 @@
 import cz.o2.proxima.core.annotations.Evolving;
 import cz.o2.proxima.core.functional.Consumer;
 import cz.o2.proxima.core.scheme.ValueSerializerFactory;
+import cz.o2.proxima.core.util.internal.ClassLoaders.ChildLayerFirstClassLoader;
 import cz.o2.proxima.internal.com.google.common.annotations.VisibleForTesting;
 import cz.o2.proxima.internal.com.google.common.collect.Streams;
 import cz.o2.proxima.typesafe.config.Config;
@@ -32,7 +33,9 @@
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.stream.Stream;
+import lombok.AccessLevel;
 import lombok.Getter;
+import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
 
 /** Repository of all entities configured in the system. */
@@ -42,6 +45,9 @@ public abstract class Repository implements Serializable {
 
   private static final long serialVersionUID = 1L;
 
+  @Setter(AccessLevel.PACKAGE)
+  private static ChildLayerFirstClassLoader jpmsClassLoader = null;
+
   /** Various validation flags. */
   public enum Validate {
     /** Do not perform any validations. */
@@ -94,6 +100,10 @@ public static Repository ofTest(Config config, Validate... validate) {
     return ConfigRepository.ofTest(config, validate);
   }
 
+  public static ClassLoader getJpmsClassloader() {
+    return jpmsClassLoader;
+  }
+
   @SuppressWarnings({"unchecked", "rawtypes"})
   private static Iterable<DataOperatorFactory<?>> readDataOperatorFactories() {
     return (Iterable) ServiceLoader.load(DataOperatorFactory.class);

direct/core/src/main/java/cz/o2/proxima/direct/core/DirectDataOperator.java

@@ -93,7 +93,9 @@ private static Factory<ExecutorService> createExecutorFactory() {
                         "Error running task in thread {}, bailing out...", thr.getName(), exc);
                     Runtime.getRuntime().exit(1);
                   });
-              t.setContextClassLoader(DirectDataOperator.class.getClassLoader());
+              if (Repository.getJpmsClassloader() != null) {
+                t.setContextClassLoader(Repository.getJpmsClassloader());
+              }
               return t;
             });
   }

direct/ingest-server/src/main/java/cz/o2/proxima/direct/server/RetrieveService.java

@@ -325,7 +325,7 @@ public void multifetch(
     for (GetRequest getRequest : request.getGetRequestList()) {
       get(
           getRequest.toBuilder().setTransactionId(request.getTransactionId()).build(),
-          new StreamObserver<GetResponse>() {
+          new StreamObserver<>() {
             @Override
             public void onNext(GetResponse getResponse) {
               response.addGetResponse(getResponse);
@@ -348,7 +348,7 @@ public void onCompleted() {}
       for (ListRequest listRequest : request.getListRequestList()) {
         listAttributes(
             listRequest.toBuilder().setTransactionId(request.getTransactionId()).build(),
-            new StreamObserver<ListResponse>() {
+            new StreamObserver<>() {
               @Override
               public void onNext(ListResponse listResponse) {
                 response.addListResponse(listResponse);