Merge branch 'master' into feature/fix-prometheus-metrics

* master:
  Improve concurrency for needed parts. (apache#3107)
  Add documents for setting up a docker based development environment (apache#3475)
  Patch to fix cppcheck with newer glibc (apache#3471)
  Make log/sink/consume Streamlet component support setName and setNumPartitions (apache#3459)

Author: thinker0

WORKSPACE

@@ -930,10 +930,12 @@ http_archive(
 
 http_archive(
     name = "com_github_danmar_cppcheck",
-    urls = ["https://github.com/danmar/cppcheck/archive/1.87.zip"],
-    strip_prefix = "cppcheck-1.87",
     build_file = "@//:third_party/cppcheck/cppcheck.BUILD",
-    sha256 = "b3de7fbdc1a23d7341b55f7f88877e106a76847bd5a07fa721c07310b625318b",
+    patch_args = ["-p2"],
+    patches = ["//third_party/cppcheck:cppcheck-readdir-fix.patch"],
+    sha256 = "cb0e66cbe2d6b655fce430cfaaa74b83ad11c91f221e3926f1ca3211bb7c906b",
+    strip_prefix = "cppcheck-1.90",
+    urls = ["https://github.com/danmar/cppcheck/archive/1.90.zip"],
 )
 
 http_archive(

examples/src/java/org/apache/heron/examples/streamlet/StreamletCloneTopology.java

@@ -144,13 +144,15 @@ public static void main(String[] args) throws Exception {
      * Elements in the first cloned streamlet go to the database sink.
      */
     splitGameScoreStreamlet.get(0)
-        .toSink(new DatabaseSink());
+        .toSink(new DatabaseSink())
+        .setName("sink0");
 
     /**
      * Elements in the second cloned streamlet go to the logging sink.
      */
     splitGameScoreStreamlet.get(1)
-        .toSink(new FormattedLogSink());
+        .toSink(new FormattedLogSink())
+        .setName("sink1");
 
     Config config = Config.defaultConfig();
 

examples/src/scala/org/apache/heron/examples/streamlet/scala/ScalaClassicalMusicTopology.scala

@@ -92,6 +92,7 @@ object ScalaClassicalMusicTopology {
       )
       .setName("joined-classical-musics-by-year")
       .log()
+      .setName("log")
 
     val config = Config.defaultConfig()
 

examples/src/scala/org/apache/heron/examples/streamlet/scala/ScalaIntegerProcessingTopology.scala

@@ -58,6 +58,8 @@ object ScalaIntegerProcessingTopology {
       .union(zeroes)
       .setName("union-of-numbers")
       .log()
+      .setName("log")
+      .setNumPartitions(1)
 
     val config = Config.newBuilder
       .setNumContainers(NUM_CONTAINERS)

heron/api/src/java/org/apache/heron/api/metric/MultiAssignableMetric.java

@@ -21,9 +21,10 @@
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 
 public class MultiAssignableMetric<T extends Number> implements IMetric<Map<String, T>> {
-  private final Map<String, AssignableMetric<T>> value = new HashMap<>();
+  private final Map<String, AssignableMetric<T>> value = new ConcurrentHashMap<>();
   private T initialValue;
 
   public MultiAssignableMetric(T initialValue) {

heron/api/src/java/org/apache/heron/api/metric/MultiCountMetric.java

@@ -21,9 +21,10 @@
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 
 public class MultiCountMetric implements IMetric<Map<String, Long>> {
-  private Map<String, CountMetric> value = new HashMap<>();
+  private Map<String, CountMetric> value = new ConcurrentHashMap<>();
 
   public MultiCountMetric() {
   }

heron/api/src/java/org/apache/heron/api/metric/MultiReducedMetric.java

@@ -21,6 +21,7 @@
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 
 /*
  * A reduce metric that can hold multiple scoped values.
@@ -29,7 +30,7 @@
  * @param <V> type of reduced value
  */
 public class MultiReducedMetric<T, U, V> implements IMetric<Map<String, V>> {
-  private Map<String, ReducedMetric<T, U, V>> value = new HashMap<>();
+  private Map<String, ReducedMetric<T, U, V>> value = new ConcurrentHashMap<>();
   private IReducer<T, U, V> reducer;
 
   public MultiReducedMetric(IReducer<T, U, V> reducer) {

heron/api/src/java/org/apache/heron/api/tuple/Fields.java

@@ -22,16 +22,16 @@
 import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.Arrays;
-import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 
 public class Fields implements Iterable<String>, Serializable {
   private static final long serialVersionUID = -1045737418722082345L;
 
   private List<String> fields;
-  private Map<String, Integer> mIndex = new HashMap<String, Integer>();
+  private Map<String, Integer> mIndex = new ConcurrentHashMap<String, Integer>();
 
   public Fields(String... pFields) {
     this(Arrays.asList(pFields));

heron/api/src/java/org/apache/heron/streamlet/Streamlet.java

@@ -45,7 +45,7 @@
  * Streamlet before doing the transformation.
  */
 @InterfaceStability.Evolving
-public interface Streamlet<R> {
+public interface Streamlet<R> extends StreamletBase<R> {
 
   /**
    * Sets the name of the BaseStreamlet.
@@ -299,21 +299,21 @@ <K> KVStreamlet<KeyedWindow<K>, Long> countByKeyAndWindow(
    * Logs every element of the streamlet using String.valueOf function
    * This is one of the sink functions in the sense that this operation returns void
    */
-  void log();
+  StreamletBase<R> log();
 
   /**
    * Applies the consumer function to every element of the stream
    * This function does not return anything.
    * @param consumer The user supplied consumer function that is invoked for each element
    * of this streamlet.
    */
-  void consume(SerializableConsumer<R> consumer);
+  StreamletBase<R> consume(SerializableConsumer<R> consumer);
 
   /**
    * Applies the sink's put function to every element of the stream
    * This function does not return anything.
    * @param sink The Sink whose put method consumes each element
    * of this streamlet.
    */
-  void toSink(Sink<R> sink);
+  StreamletBase<R> toSink(Sink<R> sink);
 }

heron/api/src/java/org/apache/heron/streamlet/StreamletBase.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.heron.streamlet;
+
+/**
+ * A Streamlet is a (potentially unbounded) ordered collection of tuples.
+ * The StreamletBase class contains basic information of a Streamlet
+ * such as name and partition count without the connection functions
+ * such as map() and filter().
+ */
+public interface StreamletBase<R> {
+
+  /**
+   * Sets the name of the BaseStreamlet.
+   * @param sName The name given by the user for this BaseStreamlet
+   * @return Returns back the Streamlet with changed name
+   */
+  StreamletBase<R> setName(String sName);
+
+  /**
+   * Gets the name of the Streamlet.
+   * @return Returns the name of the Streamlet
+   */
+  String getName();
+
+  /**
+   * Sets the number of partitions of the streamlet
+   * @param numPartitions The user assigned number of partitions
+   * @return Returns back the Streamlet with changed number of partitions
+   */
+  StreamletBase<R> setNumPartitions(int numPartitions);
+
+  /**
+   * Gets the number of partitions of this Streamlet.
+   * @return the number of partitions of this Streamlet
+   */
+  int getNumPartitions();
+
+  // This is the main interface that every Streamlet implementation should implement
+  // The main tasks are generally to make sure that appropriate names/partitions are
+  // computed and add a spout/bolt to the TopologyBuilder
+  // void build(TopologyBuilder bldr, Set<String> stageNames);
+}

heron/api/src/java/org/apache/heron/streamlet/impl/BuilderImpl.java

@@ -91,7 +91,7 @@ public TopologyBuilder build(TopologyBuilder builder) {
       streamlet.build(builder, stageNames);
     }
     for (StreamletImpl<?> streamlet : sources) {
-      if (!streamlet.allBuilt()) {
+      if (!streamlet.isFullyBuilt()) {
         throw new RuntimeException("Topology cannot be fully built! Are all sources added?");
       }
     }