Merge remote-tracking branch 'upstream/master'

Author: weisong44

samza-api/src/main/java/org/apache/samza/metadatastore/MetadataStore.java

@@ -19,8 +19,6 @@
 package org.apache.samza.metadatastore;
 
 import org.apache.samza.annotation.InterfaceStability;
-import org.apache.samza.config.Config;
-import org.apache.samza.metrics.MetricsRegistry;
 import java.util.Map;
 
 /**
@@ -32,10 +30,8 @@ public interface MetadataStore {
   /**
    * Initializes the metadata store, if applicable, setting up the underlying resources
    * and connections to the store endpoints.
-   *
-   * @param config the configuration for instantiating the MetadataStore.
    */
-  void init(Config config, MetricsRegistry metricsRegistry);
+  void init();
 
   /**
    * Gets the value associated with the specified {@code key}.

samza-core/src/main/java/org/apache/samza/container/LocalityManager.java

@@ -78,7 +78,7 @@ public LocalityManager(Config config, MetricsRegistry metricsRegistry) {
     this.config = config;
     MetadataStoreFactory metadataStoreFactory = Util.getObj(new JobConfig(config).getMetadataStoreFactory(), MetadataStoreFactory.class);
     this.metadataStore = metadataStoreFactory.getMetadataStore(SetContainerHostMapping.TYPE, config, metricsRegistry);
-    this.metadataStore.init(config, metricsRegistry);
+    this.metadataStore.init();
     this.keySerde = keySerde;
     this.valueSerde = valueSerde;
     this.taskAssignmentManager = new TaskAssignmentManager(config, metricsRegistry, keySerde, valueSerde);

samza-core/src/main/java/org/apache/samza/container/grouper/task/GroupByContainerCount.java

@@ -78,7 +78,7 @@ public Set<ContainerModel> group(Set<TaskModel> tasks) {
     // Convert to a Set of ContainerModel
     Set<ContainerModel> containerModels = new HashSet<>();
     for (int i = 0; i < containerCount; i++) {
-      containerModels.add(new ContainerModel(String.valueOf(i), i, taskGroups[i]));
+      containerModels.add(new ContainerModel(String.valueOf(i), taskGroups[i]));
     }
 
     return Collections.unmodifiableSet(containerModels);
@@ -194,7 +194,7 @@ private List<TaskGroup> getPreviousContainers(TaskAssignmentManager taskAssignme
   private void saveTaskAssignments(Set<ContainerModel> containers, TaskAssignmentManager taskAssignmentManager) {
     for (ContainerModel container : containers) {
       for (TaskName taskName : container.getTasks().keySet()) {
-        taskAssignmentManager.writeTaskContainerMapping(taskName.getTaskName(), container.getProcessorId());
+        taskAssignmentManager.writeTaskContainerMapping(taskName.getTaskName(), container.getId());
       }
     }
   }
@@ -301,7 +301,7 @@ private Set<ContainerModel> buildContainerModels(Set<TaskModel> tasks, List<Task
         containerTaskModels.put(model.getTaskName(), model);
       }
       containerModels.add(
-          new ContainerModel(container.containerId, Integer.valueOf(container.containerId), containerTaskModels));
+          new ContainerModel(container.containerId, containerTaskModels));
     }
     return Collections.unmodifiableSet(containerModels);
   }

samza-core/src/main/java/org/apache/samza/container/grouper/task/GroupByContainerIds.java

@@ -100,9 +100,7 @@ public Set<ContainerModel> group(Set<TaskModel> tasks, List<String> containersId
     // Convert to a Set of ContainerModel
     Set<ContainerModel> containerModels = new HashSet<>();
     for (int i = 0; i < containerCount; i++) {
-      // containerId in ContainerModel constructor is set to -1 because processorId can be any string and does
-      // not have an integer equivalent. So, we set it to -1. After 0.13, this parameter will be removed.
-      containerModels.add(new ContainerModel(containersIds.get(i), -1, taskGroups[i]));
+      containerModels.add(new ContainerModel(containersIds.get(i), taskGroups[i]));
     }
 
     return Collections.unmodifiableSet(containerModels);

samza-core/src/main/java/org/apache/samza/container/grouper/task/SingleContainerGrouperFactory.java

@@ -50,7 +50,7 @@ public Set<ContainerModel> group(Set<TaskModel> taskModels) {
     for (TaskModel taskModel: taskModels) {
       taskNameTaskModelMap.put(taskModel.getTaskName(), taskModel);
     }
-    ContainerModel containerModel = new ContainerModel(containerId, -1, taskNameTaskModelMap);
+    ContainerModel containerModel = new ContainerModel(containerId, taskNameTaskModelMap);
     return Collections.singleton(containerModel);
   }
 }

samza-core/src/main/java/org/apache/samza/container/grouper/task/TaskAssignmentManager.java

@@ -84,7 +84,7 @@ public TaskAssignmentManager(Config config, MetricsRegistry metricsRegistry, Ser
   }
 
   public void init(Config config, MetricsRegistry metricsRegistry) {
-    this.metadataStore.init(config, metricsRegistry);
+    this.metadataStore.init();
   }
 
   /**

samza-core/src/main/java/org/apache/samza/coordinator/metadatastore/CoordinatorStreamStore.java

@@ -89,7 +89,7 @@ public CoordinatorStreamStore(String namespace, Config config, MetricsRegistry m
   }
 
   @Override
-  public void init(Config config, MetricsRegistry metricsRegistry) {
+  public void init() {
     if (isInitialized.compareAndSet(false, true)) {
       LOG.info("Starting the coordinator stream system consumer with config: {}.", config);
       registerConsumer();

samza-core/src/main/java/org/apache/samza/job/model/ContainerModel.java

@@ -19,10 +19,9 @@
 
 package org.apache.samza.job.model;
 
-import org.apache.samza.container.TaskName;
-
 import java.util.Collections;
 import java.util.Map;
+import org.apache.samza.container.TaskName;
 
 /**
  * <p>
@@ -41,28 +40,16 @@
  * </p>
  */
 public class ContainerModel {
-  @Deprecated
-  private final int containerId;
-  private final String processorId;
+  private final String id;
   private final Map<TaskName, TaskModel> tasks;
 
-  public ContainerModel(String processorId, int containerId, Map<TaskName, TaskModel> tasks) {
-    this.containerId = containerId;
-    if (processorId == null) {
-      this.processorId = String.valueOf(containerId);
-    } else {
-      this.processorId = processorId;
-    }
+  public ContainerModel(String id, Map<TaskName, TaskModel> tasks) {
+    this.id = id;
     this.tasks = Collections.unmodifiableMap(tasks);
   }
 
-  @Deprecated
-  public int getContainerId() {
-    return containerId;
-  }
-
-  public String getProcessorId() {
-    return processorId;
+  public String getId() {
+    return id;
   }
 
   public Map<TaskName, TaskModel> getTasks() {
@@ -71,14 +58,14 @@ public Map<TaskName, TaskModel> getTasks() {
 
   @Override
   public String toString() {
-    return "ContainerModel [processorId=" + processorId + ", tasks=" + tasks + "]";
+    return "ContainerModel [id=" + id + ", tasks=" + tasks + "]";
   }
 
   @Override
   public int hashCode() {
     final int prime = 31;
     int result = 1;
-    result = prime * result + ((processorId == null) ? 0 : processorId.hashCode());
+    result = prime * result + ((id == null) ? 0 : id.hashCode());
     result = prime * result + ((tasks == null) ? 0 : tasks.hashCode());
     return result;
   }
@@ -92,7 +79,7 @@ public boolean equals(Object obj) {
     if (getClass() != obj.getClass())
       return false;
     ContainerModel other = (ContainerModel) obj;
-    if (!processorId.equals(other.processorId))
+    if (!id.equals(other.id))
       return false;
     if (tasks == null) {
       if (other.tasks != null)

samza-core/src/main/java/org/apache/samza/job/model/JobModel.java

@@ -25,7 +25,6 @@
 import org.apache.samza.config.Config;
 import org.apache.samza.container.LocalityManager;
 import org.apache.samza.coordinator.stream.messages.SetContainerHostMapping;
-import org.codehaus.jackson.annotate.JsonIgnoreProperties;
 
 /**
  * <p>
@@ -39,7 +38,6 @@
  * an id, partition information, etc.
  * </p>
  */
-@JsonIgnoreProperties(ignoreUnknown = true)
 public class JobModel {
   private static final String EMPTY_STRING = "";
   private final Config config;

samza-core/src/main/java/org/apache/samza/serializers/model/JsonContainerModelMixIn.java

@@ -19,29 +19,35 @@
 
 package org.apache.samza.serializers.model;
 
+import java.util.Map;
 import org.apache.samza.container.TaskName;
 import org.apache.samza.job.model.TaskModel;
-import org.codehaus.jackson.annotate.JsonCreator;
 import org.codehaus.jackson.annotate.JsonProperty;
 
-import java.util.Map;
-
 /**
- * A mix-in Jackson class to convert Samza's ContainerModel to/from JSON.
+ * A mix-in Jackson class to convert {@link org.apache.samza.job.model.ContainerModel} to JSON.
+ * Notes:
+ * 1) Constructor is not needed because this mixin is not used for deserialization. See {@link SamzaObjectMapper}.
+ * 2) It is unnecessary to use {@link org.codehaus.jackson.annotate.JsonIgnoreProperties#ignoreUnknown()} here since
+ * {@link SamzaObjectMapper} already uses custom deserialization code for the
+ * {@link org.apache.samza.job.model.ContainerModel}.
+ * 3) See {@link SamzaObjectMapper} for more context about why the JSON keys are named in this specified way.
  */
 public abstract class JsonContainerModelMixIn {
-  @JsonCreator
-  public JsonContainerModelMixIn(@JsonProperty("processor-id") String processorId, @JsonProperty("tasks") Map<TaskName, TaskModel> tasks) {
-  }
-
-  @Deprecated
-  @JsonProperty("container-id")
-  abstract int getContainerId();
-
-  @JsonProperty("processor-id")
-  abstract String getProcessorId();
-
-  @JsonProperty("tasks")
+  /**
+   * This is intentionally not "id" for backwards compatibility reasons. See {@link SamzaObjectMapper} for more details.
+   */
+  static final String PROCESSOR_ID_KEY = "processor-id";
+  /**
+   * This is used for backwards compatibility. See {@link SamzaObjectMapper} for more details.
+   */
+  static final String CONTAINER_ID_KEY = "container-id";
+  static final String TASKS_KEY = "tasks";
+
+  @JsonProperty(PROCESSOR_ID_KEY)
+  abstract String getId();
+
+  @JsonProperty(TASKS_KEY)
   abstract Map<TaskName, TaskModel> getTasks();
 }