Merge remote-tracking branch 'upstream/master'

Author: weisong44

samza-api/src/main/java/org/apache/samza/config/Config.java

@@ -154,6 +154,9 @@ public List<String> getList(String k, List<String> defaultValue) {
       return defaultValue;
 
     String value = get(k);
+    if (value.trim().isEmpty()) {
+      return defaultValue;
+    }
     String[] pieces = value.split("\\s*,\\s*");
     return Arrays.asList(pieces);
   }

samza-api/src/main/java/org/apache/samza/metrics/ListGauge.java

@@ -109,8 +109,9 @@ public void visit(MetricsVisitor visitor) {
 
   /**
    * Evicts entries from the elements list, based on the given item-size and durationThreshold.
+   * Concurrent eviction threads can cause incorrectness (when reading elements.size or elements.peek).
    */
-  private void evict() {
+  private synchronized void evict() {
     this.evictBasedOnSize();
     this.evictBasedOnTimestamp();
   }

samza-api/src/main/java/org/apache/samza/runtime/LocationId.java

@@ -0,0 +1,60 @@
+/*
+ * 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.samza.runtime;
+
+import java.util.Objects;
+
+/**
+ * Represents the physical execution environment of the StreamProcessor.
+ * All the stream processors which run from a LocationId should be able to share (read/write)
+ * their local state stores.
+ */
+public class LocationId {
+  private final String locationId;
+
+  public LocationId(String locationId) {
+    if (locationId == null) {
+      throw new IllegalArgumentException("LocationId cannot be null");
+    }
+    this.locationId = locationId;
+  }
+
+  public String getId() {
+    return this.locationId;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) return true;
+    if (o == null || getClass() != o.getClass()) return false;
+    LocationId that = (LocationId) o;
+
+    return Objects.equals(locationId, that.locationId);
+  }
+
+  @Override
+  public int hashCode() {
+    return locationId.hashCode();
+  }
+
+  @Override
+  public String toString() {
+    return locationId;
+  }
+}

samza-core/src/main/java/org/apache/samza/table/remote/CreditFunction.java renamed to samza-api/src/main/java/org/apache/samza/runtime/LocationIdProvider.java

@@ -16,21 +16,13 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-
-package org.apache.samza.table.remote;
-
-import java.io.Serializable;
-import java.util.function.Function;
-
-import org.apache.samza.operators.KV;
-
+package org.apache.samza.runtime;
 
 /**
- * Function interface for providing rate limiting credits for each table record.
- * This interface allows callers to pass in lambda expressions which are otherwise
- * non-serializable as-is.
- * @param <K> the type of the key
- * @param <V> the type of the value
+ * Generates {@link LocationId} that uniquely identifies the
+ * execution environment of a stream processor.
  */
-public interface CreditFunction<K, V> extends Function<KV<K, V>, Integer>, Serializable {
-}
+public interface LocationIdProvider {
+
+  LocationId getLocationId();
+}

samza-api/src/main/java/org/apache/samza/runtime/LocationIdProviderFactory.java

@@ -0,0 +1,28 @@
+/*
+ * 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.samza.runtime;
+
+import org.apache.samza.config.Config;
+
+/**
+ * Builds the {@link LocationIdProvider}.
+ */
+public interface LocationIdProviderFactory {
+  LocationIdProvider getLocationIdProvider(Config config);
+}

samza-api/src/main/java/org/apache/samza/table/ReadWriteTable.java

@@ -19,6 +19,8 @@
 package org.apache.samza.table;
 
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
 import org.apache.samza.annotation.InterfaceStability;
 import org.apache.samza.storage.kv.Entry;
 
@@ -33,7 +35,8 @@
 public interface ReadWriteTable<K, V> extends ReadableTable<K, V> {
 
   /**
-   * Updates the mapping of the specified key-value pair; Associates the specified {@code key} with the specified {@code value}.
+   * Updates the mapping of the specified key-value pair;
+   * Associates the specified {@code key} with the specified {@code value}.
    *
    * The key is deleted from the table if value is {@code null}.
    *
@@ -43,6 +46,18 @@ public interface ReadWriteTable<K, V> extends ReadableTable<K, V> {
    */
   void put(K key, V value);
 
+  /**
+   * Asynchronously updates the mapping of the specified key-value pair;
+   * Associates the specified {@code key} with the specified {@code value}.
+   * The key is deleted from the table if value is {@code null}.
+   *
+   * @param key the key with which the specified {@code value} is to be associated.
+   * @param value the value with which the specified {@code key} is to be associated.
+   * @throws NullPointerException if the specified {@code key} is {@code null}.
+   * @return CompletableFuture for the operation
+   */
+  CompletableFuture<Void> putAsync(K key, V value);
+
   /**
    * Updates the mappings of the specified key-value {@code entries}.
    *
@@ -53,6 +68,16 @@ public interface ReadWriteTable<K, V> extends ReadableTable<K, V> {
    */
   void putAll(List<Entry<K, V>> entries);
 
+  /**
+   * Asynchronously updates the mappings of the specified key-value {@code entries}.
+   * A key is deleted from the table if its corresponding value is {@code null}.
+   *
+   * @param entries the updated mappings to put into this table.
+   * @throws NullPointerException if any of the specified {@code entries} has {@code null} as key.
+   * @return CompletableFuture for the operation
+   */
+  CompletableFuture<Void> putAllAsync(List<Entry<K, V>> entries);
+
   /**
    * Deletes the mapping for the specified {@code key} from this table (if such mapping exists).
    *
@@ -61,6 +86,14 @@ public interface ReadWriteTable<K, V> extends ReadableTable<K, V> {
    */
   void delete(K key);
 
+  /**
+   * Asynchronously deletes the mapping for the specified {@code key} from this table (if such mapping exists).
+   * @param key the key for which the mapping is to be deleted.
+   * @throws NullPointerException if the specified {@code key} is {@code null}.
+   * @return CompletableFuture for the operation
+   */
+  CompletableFuture<Void> deleteAsync(K key);
+
   /**
    * Deletes the mappings for the specified {@code keys} from this table.
    *
@@ -69,10 +102,16 @@ public interface ReadWriteTable<K, V> extends ReadableTable<K, V> {
    */
   void deleteAll(List<K> keys);
 
+  /**
+   * Asynchronously deletes the mappings for the specified {@code keys} from this table.
+   * @param keys the keys for which the mappings are to be deleted.
+   * @throws NullPointerException if the specified {@code keys} list, or any of the keys, is {@code null}.
+   * @return CompletableFuture for the operation
+   */
+  CompletableFuture<Void> deleteAllAsync(List<K> keys);
 
   /**
    * Flushes the underlying store of this table, if applicable.
    */
   void flush();
-
 }

samza-api/src/main/java/org/apache/samza/table/ReadableTable.java

@@ -20,6 +20,7 @@
 
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.CompletableFuture;
 
 import org.apache.samza.annotation.InterfaceStability;
 import org.apache.samza.container.SamzaContainerContext;
@@ -54,6 +55,15 @@ default void init(SamzaContainerContext containerContext, TaskContext taskContex
    */
   V get(K key);
 
+  /**
+   * Asynchronously gets the value associated with the specified {@code key}.
+   *
+   * @param key the key with which the associated value is to be fetched.
+   * @return completableFuture for the requested value
+   * @throws NullPointerException if the specified {@code key} is {@code null}.
+   */
+  CompletableFuture<V> getAsync(K key);
+
   /**
    * Gets the values with which the specified {@code keys} are associated.
    *
@@ -63,6 +73,15 @@ default void init(SamzaContainerContext containerContext, TaskContext taskContex
    */
   Map<K, V> getAll(List<K> keys);
 
+  /**
+   * Asynchronously gets the values with which the specified {@code keys} are associated.
+   *
+   * @param keys the keys with which the associated values are to be fetched.
+   * @return completableFuture for the requested entries
+   * @throws NullPointerException if the specified {@code keys} list, or any of the keys, is {@code null}.
+   */
+  CompletableFuture<Map<K, V>> getAllAsync(List<K> keys);
+
   /**
    * Close the table and release any resources acquired
    */

samza-api/src/test/java/org/apache/samza/config/TestConfig.java

@@ -19,6 +19,8 @@
 
 package org.apache.samza.config;
 
+import java.util.Collections;
+import java.util.List;
 import org.junit.Test;
 
 import java.util.HashMap;
@@ -85,4 +87,43 @@ public void testSanitize() {
     assertEquals(Config.SENSITIVE_MASK, sanitized.get("sensitive.key3"));
     assertEquals(Config.SENSITIVE_MASK, sanitized.get("sensitive.key4"));
   }
+
+  @Test
+  public void testGetList() {
+    Map<String, String> m = new HashMap<String, String>() {
+      {
+        put("key1", " ");
+        put("key2", "");
+        put("key3", "  value1  ");
+        put("key4", "value1,value2");
+        put("key5", "value1, value2");
+        put("key6", "value1  ,   value2");
+      }
+    };
+
+    Config config = new MapConfig(m);
+    List<String> list = config.getList("key1", Collections.<String>emptyList());
+    assertEquals(0, list.size());
+
+    list = config.getList("key2", Collections.<String>emptyList());
+    assertEquals(0, list.size());
+
+    list = config.getList("key3");
+    assertEquals("  value1  ", list.get(0));
+
+    list = config.getList("key4");
+    assertEquals("value1", list.get(0));
+    assertEquals("value2", list.get(1));
+
+    list = config.getList("key5");
+    assertEquals("value1", list.get(0));
+    assertEquals("value2", list.get(1));
+
+    list = config.getList("key6");
+    assertEquals("value1", list.get(0));
+    assertEquals("value2", list.get(1));
+
+    list = config.getList("UndefinedKey", Collections.<String>emptyList());
+    assertEquals(0, list.size());
+  }
 }

samza-core/src/main/java/org/apache/samza/execution/ExecutionPlanner.java

@@ -28,7 +28,6 @@
 import java.util.Map;
 import java.util.Queue;
 import java.util.Set;
-
 import org.apache.samza.SamzaException;
 import org.apache.samza.config.ApplicationConfig;
 import org.apache.samza.config.ClusterManagerConfig;
@@ -186,11 +185,12 @@ private void validateConfig() {
     // The visited set keeps track of the join specs that have been already inserted in the queue before
     Set<OperatorSpec> visited = new HashSet<>();
 
-    jobGraph.getSpecGraph().getInputOperators().entrySet().forEach(entry -> {
-        StreamConfig streamConfig = new StreamConfig(config);
-        StreamEdge streamEdge = jobGraph.getOrCreateStreamEdge(getStreamSpec(entry.getKey(), streamConfig));
+    StreamConfig streamConfig = new StreamConfig(config);
+
+    jobGraph.getSpecGraph().getInputOperators().forEach((key, value) -> {
+        StreamEdge streamEdge = jobGraph.getOrCreateStreamEdge(getStreamSpec(key, streamConfig));
         // Traverses the StreamGraph to find and update mappings for all Joins reachable from this input StreamEdge
-        findReachableJoins(entry.getValue(), streamEdge, joinSpecToStreamEdges, streamEdgeToJoinSpecs, joinQ, visited);
+        findReachableJoins(value, streamEdge, joinSpecToStreamEdges, streamEdgeToJoinSpecs, joinQ, visited);
       });
 
     // At this point, joinQ contains joinSpecs where at least one of the input stream edge partitions is known.
@@ -204,16 +204,21 @@ private void validateConfig() {
           if (partitions == StreamEdge.PARTITIONS_UNKNOWN) {
             //if the partition is not assigned
             partitions = edgePartitions;
+            log.info("Inferred the partition count {} for the join operator {} from {}."
+                , new Object[] {partitions, join.getOpId(), edge.getName()});
           } else if (partitions != edgePartitions) {
             throw  new SamzaException(String.format(
-                "Unable to resolve input partitions of stream %s for join. Expected: %d, Actual: %d",
-                edge.getName(), partitions, edgePartitions));
+                "Unable to resolve input partitions of stream %s for the join %s. Expected: %d, Actual: %d",
+                edge.getName(), join.getOpId(), partitions, edgePartitions));
           }
         }
       }
+
       // assign the partition count for intermediate streams
       for (StreamEdge edge : joinSpecToStreamEdges.get(join)) {
         if (edge.getPartitionCount() <= 0) {
+          log.info("Set the partition count to {} for input stream {} to the join {}.",
+              new Object[] {partitions, edge.getName(), join.getOpId()});
           edge.setPartitionCount(partitions);
 
           // find other joins can be inferred by setting this edge
@@ -278,6 +283,7 @@ private static void calculateIntStreamPartitions(JobGraph jobGraph, Config confi
     }
     for (StreamEdge edge : jobGraph.getIntermediateStreamEdges()) {
       if (edge.getPartitionCount() <= 0) {
+        log.info("Set the partition count for intermediate stream {} to {}.", edge.getName(), partitions);
         edge.setPartitionCount(partitions);
       }
     }