[REEF-2024] Prepare existing artifacts for new bridge module. (#1467)

The new bridge needs access to create an EvaluatorDescriptor. This PR creates a factor and builder for EvaluatorDescriptor that can be injected via Tang. Default implementation creates an EvaluatorDescriptImpl.

Pull Request:
   Closes #1467

Author: tcondie

lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/evaluator/EvaluatorDescriptorBuilder.java

@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.reef.runtime.common.driver.evaluator;
+
+import org.apache.reef.driver.catalog.NodeDescriptor;
+import org.apache.reef.driver.evaluator.EvaluatorDescriptor;
+import org.apache.reef.driver.evaluator.EvaluatorProcess;
+
+/**
+ * A builder for evaluator descriptors.
+ */
+public interface EvaluatorDescriptorBuilder extends org.apache.reef.util.Builder<EvaluatorDescriptor> {
+
+  /**
+   * Set the node descriptor for this evaluator.
+   * @param nodeDescriptor for this evaluator
+   * @return this
+   */
+  EvaluatorDescriptorBuilder setNodeDescriptor(final NodeDescriptor nodeDescriptor);
+
+  /**
+   * Amount of memory dedicated to this evaluator.
+   * @param megaBytes of dedicated memory
+   * @return this
+   */
+  EvaluatorDescriptorBuilder setMemory(final int megaBytes);
+
+  /**
+   * Set the number of cores.
+   * @param numberOfCores dedicated for this evaluator
+   * @return this
+   */
+  EvaluatorDescriptorBuilder setNumberOfCores(final int numberOfCores);
+
+  /**
+   * The process used to run this evaluator.
+   * @param evaluatorProcess for this evaluator
+   * @return this
+   */
+  EvaluatorDescriptorBuilder setEvaluatorProcess(final EvaluatorProcess evaluatorProcess);
+
+  /**
+   * The runtime name for this evaluator.
+   * @param runtimeName for this evaluator
+   * @return this
+   */
+  EvaluatorDescriptorBuilder setRuntimeName(final String runtimeName);
+}

lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/evaluator/EvaluatorDescriptorBuilderFactory.java

@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.reef.runtime.common.driver.evaluator;
+
+import org.apache.reef.driver.evaluator.EvaluatorDescriptor;
+import org.apache.reef.tang.annotations.DefaultImplementation;
+
+/**
+ * Evaluator descriptor builder factory interface.
+ */
+@DefaultImplementation(EvaluatorDescriptorImpl.BuilderFactory.class)
+public interface EvaluatorDescriptorBuilderFactory {
+  /**
+   * Create a new evaluator descriptor builder.
+   * @return new evaluator descriptor builder
+   */
+  EvaluatorDescriptorBuilder newBuilder();
+
+  /**
+   * Create an evaluator descriptor builder that is initialized to another copy.
+   * @param copy to initialize builder
+   * @return evaluator descriptor builder initialized to copy
+   */
+  EvaluatorDescriptorBuilder newBuilder(final EvaluatorDescriptor copy);
+}

lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/evaluator/EvaluatorDescriptorImpl.java

@@ -24,6 +24,8 @@
 import org.apache.reef.driver.evaluator.EvaluatorDescriptor;
 import org.apache.reef.driver.evaluator.EvaluatorProcess;
 
+import javax.inject.Inject;
+
 /**
  * A simple all-data implementation of EvaluatorDescriptor.
  */
@@ -37,11 +39,12 @@ final class EvaluatorDescriptorImpl implements EvaluatorDescriptor {
   private EvaluatorProcess process;
   private final String runtimeName;
 
-  EvaluatorDescriptorImpl(final NodeDescriptor nodeDescriptor,
-                          final int megaBytes,
-                          final int numberOfCores,
-                          final EvaluatorProcess process,
-                          final String runtimeName) {
+  private EvaluatorDescriptorImpl(
+      final NodeDescriptor nodeDescriptor,
+      final int megaBytes,
+      final int numberOfCores,
+      final EvaluatorProcess process,
+      final String runtimeName) {
     this.nodeDescriptor = nodeDescriptor;
     this.megaBytes = megaBytes;
     this.numberOfCores = numberOfCores;
@@ -80,4 +83,84 @@ public int getNumberOfCores() {
   public String getRuntimeName() {
     return this.runtimeName;
   }
+
+  /**
+   * Evaluator descriptor builder factory that creates a  new evaluator descriptor builder impl.
+   */
+  static final class BuilderFactory implements EvaluatorDescriptorBuilderFactory {
+    @Inject
+    private BuilderFactory() {
+    }
+
+    @Override
+    public EvaluatorDescriptorBuilder newBuilder() {
+      return new Builder();
+    }
+
+    @Override
+    public EvaluatorDescriptorBuilder newBuilder(final EvaluatorDescriptor copy) {
+      return newBuilder()
+          .setNodeDescriptor(copy.getNodeDescriptor())
+          .setMemory(copy.getMemory())
+          .setNumberOfCores(copy.getNumberOfCores())
+          .setEvaluatorProcess(copy.getProcess())
+          .setRuntimeName(copy.getRuntimeName());
+    }
+  }
+
+  /**
+   * An builder for this evaluator descriptor implementation.
+   */
+  private static final class Builder implements EvaluatorDescriptorBuilder {
+    private NodeDescriptor nodeDescriptor = null;
+    private int memory = 0;
+    private int numberOfCores = 0;
+    private EvaluatorProcess evaluatorProcess = null;
+    private String runtimeName = null;
+
+    @Override
+    public Builder setNodeDescriptor(final NodeDescriptor nodeDescriptor) {
+      this.nodeDescriptor = nodeDescriptor;
+      return this;
+    }
+
+    @Override
+    public Builder setMemory(final int megaBytes) {
+      this.memory = megaBytes;
+      return this;
+    }
+
+    @Override
+    public Builder setNumberOfCores(final int numberOfCores) {
+      this.numberOfCores = numberOfCores;
+      return this;
+    }
+
+    @Override
+    public Builder setEvaluatorProcess(final EvaluatorProcess evaluatorProcess) {
+      this.evaluatorProcess = evaluatorProcess;
+      return this;
+    }
+
+    @Override
+    public Builder setRuntimeName(final String runtimeName) {
+      this.runtimeName = runtimeName;
+      return this;
+    }
+
+    @Override
+    public EvaluatorDescriptor build() {
+      if (this.memory == 0) {
+        throw new IllegalArgumentException("memory not set");
+      } else if (this.numberOfCores == 0) {
+        throw new IllegalArgumentException("number of cores not set");
+      }
+      return new EvaluatorDescriptorImpl(
+          this.nodeDescriptor,
+          this.memory,
+          this.numberOfCores,
+          this.evaluatorProcess,
+          this.runtimeName);
+    }
+  }
 }

lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/evaluator/EvaluatorManager.java

@@ -91,7 +91,7 @@ public final class EvaluatorManager implements Identifiable, AutoCloseable {
   private final ResourceReleaseHandler resourceReleaseHandler;
   private final ResourceLaunchHandler resourceLaunchHandler;
   private final String evaluatorId;
-  private final EvaluatorDescriptorImpl evaluatorDescriptor;
+  private final EvaluatorDescriptorBuilderFactory evaluatorDescriptorBuilderFactory;
   private final ContextRepresenters contextRepresenters;
   private final EvaluatorMessageDispatcher messageDispatcher;
   private final EvaluatorControlHandler evaluatorControlHandler;
@@ -107,14 +107,15 @@ public final class EvaluatorManager implements Identifiable, AutoCloseable {
   private final EvaluatorIdlenessThreadPool idlenessThreadPool;
 
   // Mutable fields
+  private EvaluatorDescriptor evaluatorDescriptor;
   private Optional<TaskRepresenter> task = Optional.empty();
   private boolean resourceNotReleased = true;
   private boolean allocationNotFired = true;
 
   @Inject
   private EvaluatorManager(
       @Parameter(EvaluatorIdentifier.class) final String evaluatorId,
-      @Parameter(EvaluatorDescriptorName.class) final EvaluatorDescriptorImpl evaluatorDescriptor,
+      @Parameter(EvaluatorDescriptorName.class) final EvaluatorDescriptor evaluatorDescriptor,
       @Parameter(EvaluatorConfigurationProviders.class)
         final Set<ConfigurationProvider> evaluatorConfigurationProviders,
       final Clock clock,
@@ -131,12 +132,14 @@ private EvaluatorManager(
       final EventHandlerIdlenessSource idlenessSource,
       final LoggingScopeFactory loggingScopeFactory,
       final DriverRestartManager driverRestartManager,
-      final EvaluatorIdlenessThreadPool idlenessThreadPool) {
+      final EvaluatorIdlenessThreadPool idlenessThreadPool,
+      final EvaluatorDescriptorBuilderFactory evaluatorDescriptorBuilderFactory) {
 
     LOG.log(Level.FINEST, "Instantiating 'EvaluatorManager' for evaluator: {0}", evaluatorId);
 
     this.evaluatorId = evaluatorId;
     this.evaluatorDescriptor = evaluatorDescriptor;
+    this.evaluatorDescriptorBuilderFactory = evaluatorDescriptorBuilderFactory;
     this.evaluatorConfigurationProviders = evaluatorConfigurationProviders;
 
     this.clock = clock;
@@ -209,7 +212,9 @@ public String getId() {
   }
 
   public void setProcess(final EvaluatorProcess process) {
-    this.evaluatorDescriptor.setProcess(process);
+    this.evaluatorDescriptor = this.evaluatorDescriptorBuilderFactory.newBuilder(this.evaluatorDescriptor)
+        .setEvaluatorProcess(process)
+        .build();
   }
 
   public EvaluatorDescriptor getEvaluatorDescriptor() {
@@ -708,6 +713,6 @@ public static final class EvaluatorIdentifier implements Name<String> {
    * The Evaluator Host.
    */
   @NamedParameter(doc = "The Evaluator Host.")
-  public static final class EvaluatorDescriptorName implements Name<EvaluatorDescriptorImpl> {
+  public static final class EvaluatorDescriptorName implements Name<EvaluatorDescriptor> {
   }
 }

lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/evaluator/EvaluatorManagerFactory.java

@@ -23,6 +23,7 @@
 import org.apache.reef.annotations.audience.Private;
 import org.apache.reef.driver.catalog.NodeDescriptor;
 import org.apache.reef.driver.catalog.ResourceCatalog;
+import org.apache.reef.driver.evaluator.EvaluatorDescriptor;
 import org.apache.reef.driver.evaluator.EvaluatorProcessFactory;
 import org.apache.reef.runtime.common.driver.catalog.ResourceCatalogImpl;
 import org.apache.reef.runtime.common.driver.resourcemanager.*;
@@ -45,14 +46,17 @@ public final class EvaluatorManagerFactory {
   private final Injector injector;
   private final ResourceCatalog resourceCatalog;
   private final EvaluatorProcessFactory processFactory;
+  private final EvaluatorDescriptorBuilderFactory evaluatorDescriptorBuilderFactory;
 
   @Inject
   EvaluatorManagerFactory(final Injector injector,
                           final ResourceCatalog resourceCatalog,
-                          final EvaluatorProcessFactory processFactory) {
+                          final EvaluatorProcessFactory processFactory,
+                          final EvaluatorDescriptorBuilderFactory evaluatorDescriptorBuilderFactory) {
     this.injector = injector;
     this.resourceCatalog = resourceCatalog;
     this.processFactory = processFactory;
+    this.evaluatorDescriptorBuilderFactory = evaluatorDescriptorBuilderFactory;
   }
 
   private EvaluatorManager getNewEvaluatorManagerInstanceForResource(
@@ -72,10 +76,13 @@ private EvaluatorManager getNewEvaluatorManagerInstanceForResource(
       ((ResourceCatalogImpl) resourceCatalog).handle(nodeDescriptorEvent);
       nodeDescriptor = this.resourceCatalog.getNode(nodeId);
     }
-    final EvaluatorDescriptorImpl evaluatorDescriptor = new EvaluatorDescriptorImpl(nodeDescriptor,
-        resourceEvent.getResourceMemory(), resourceEvent.getVirtualCores().get(),
-        processFactory.newEvaluatorProcess(), resourceEvent.getRuntimeName());
-
+    final EvaluatorDescriptor evaluatorDescriptor = evaluatorDescriptorBuilderFactory.newBuilder()
+        .setNodeDescriptor(nodeDescriptor)
+        .setMemory(resourceEvent.getResourceMemory())
+        .setNumberOfCores(resourceEvent.getVirtualCores().get())
+        .setEvaluatorProcess(processFactory.newEvaluatorProcess())
+        .setRuntimeName(resourceEvent.getRuntimeName())
+        .build();
     LOG.log(Level.FINEST, "Resource allocation: new evaluator id[{0}]", resourceEvent.getIdentifier());
     final EvaluatorManager evaluatorManager =
         getNewEvaluatorManagerInstance(resourceEvent.getIdentifier(), evaluatorDescriptor);
@@ -90,7 +97,7 @@ private EvaluatorManager getNewEvaluatorManagerInstanceForResource(
    * @param desc NodeDescriptor on which the Evaluator executes.
    * @return a new EvaluatorManager instance.
    */
-  private EvaluatorManager getNewEvaluatorManagerInstance(final String id, final EvaluatorDescriptorImpl desc) {
+  private EvaluatorManager getNewEvaluatorManagerInstance(final String id, final EvaluatorDescriptor desc) {
     LOG.log(Level.FINEST, "Creating Evaluator Manager for Evaluator ID {0}", id);
     final Injector child = this.injector.forkInjector();
 
@@ -134,8 +141,12 @@ public EvaluatorManager getNewEvaluatorManagerForNewEvaluator(
   public EvaluatorManager getNewEvaluatorManagerForEvaluatorFailedDuringDriverRestart(
       final ResourceStatusEvent resourceStatusEvent) {
     return getNewEvaluatorManagerInstance(resourceStatusEvent.getIdentifier(),
-        new EvaluatorDescriptorImpl(null, 128, 1, processFactory.newEvaluatorProcess(),
-                resourceStatusEvent.getRuntimeName()));
+        this.evaluatorDescriptorBuilderFactory.newBuilder()
+            .setMemory(128)
+            .setNumberOfCores(1)
+            .setEvaluatorProcess(processFactory.newEvaluatorProcess())
+            .setRuntimeName(resourceStatusEvent.getRuntimeName())
+            .build());
   }
 
   /**