Merge branch 'master' of https://github.com/apache/samza

Author: Boris Shkolnik

build.gradle

@@ -89,6 +89,7 @@ rat {
     'samza-hdfs/src/main/resources/**',
     'samza-hdfs/src/test/resources/**',
     'out/**',
+    'hs_err_pid*.log',
     'state/**'
   ]
 }

samza-api/src/main/java/org/apache/samza/storage/kv/KeyValueStore.java

@@ -46,7 +46,19 @@ public interface KeyValueStore<K, V> {
    * @return a map of the keys that were found and their respective values.
    * @throws NullPointerException if the specified {@code keys} list, or any of the keys, is {@code null}.
    */
-  Map<K, V> getAll(List<K> keys);
+  default Map<K, V> getAll(List<K> keys) {
+    Map<K, V> map = new HashMap<>(keys.size());
+
+    for (K key : keys) {
+      V value = get(key);
+
+      if (value != null) {
+        map.put(key, value);
+      }
+    }
+
+    return map;
+  }
 
   /**
    * Updates the mapping of the specified key-value pair; Associates the specified {@code key} with the specified {@code value}.
@@ -79,7 +91,11 @@ public interface KeyValueStore<K, V> {
    * @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}.
    */
-  void deleteAll(List<K> keys);
+  default void deleteAll(List<K> keys) {
+    for (K key : keys) {
+      delete(key);
+    }
+  }
 
   /**
    * Returns an iterator for a sorted range of entries specified by [{@code from}, {@code to}).
@@ -111,53 +127,4 @@ public interface KeyValueStore<K, V> {
    * Flushes this key-value store, if applicable.
    */
   void flush();
-
-  /**
-   * Represents an extension for classes that implement {@link KeyValueStore}.
-   */
-  // TODO replace with default interface methods when we can use Java 8 features.
-  class Extension {
-    private Extension() {
-      // This class cannot be instantiated
-    }
-
-    /**
-     * Gets the values with which the specified {@code keys} are associated.
-     *
-     * @param store the key-value store for which this operation is to be performed.
-     * @param keys the keys with which the associated values are to be fetched.
-     * @param <K> the type of keys maintained by the specified {@code store}.
-     * @param <V> the type of values maintained by the specified {@code store}.
-     * @return a map of the keys that were found and their respective values.
-     * @throws NullPointerException if the specified {@code keys} list, or any of the keys, is {@code null}.
-     */
-    public static <K, V> Map<K, V> getAll(final KeyValueStore<K, V> store, final List<K> keys) {
-      final Map<K, V> map = new HashMap<>(keys.size());
-
-      for (final K key : keys) {
-        final V value = store.get(key);
-
-        if (value != null) {
-          map.put(key, value);
-        }
-      }
-
-      return map;
-    }
-
-    /**
-     * Deletes the mappings for the specified {@code keys} from this key-value store (if such mappings exist).
-     *
-     * @param store the key-value store for which this operation is to be performed.
-     * @param keys the keys for which the mappings are to be deleted.
-     * @param <K> the type of keys maintained by the specified {@code store}.
-     * @param <V> the type of values maintained by the specified {@code store}.
-     * @throws NullPointerException if the specified {@code keys} list, or any of the keys, is {@code null}.
-     */
-    public static <K, V> void deleteAll(final KeyValueStore<K, V> store, final List<K> keys) {
-      for (final K key : keys) {
-        store.delete(key);
-      }
-    }
-  }
 }

samza-core/src/main/java/org/apache/samza/task/StreamOperatorTask.java

@@ -41,7 +41,7 @@
  * A {@link StreamTask} implementation that brings all the operator API implementation components together and
  * feeds the input messages into the user-defined transformation chains in {@link StreamApplication}.
  */
-public final class StreamOperatorTask implements StreamTask, InitableTask, WindowableTask, ClosableTask {
+public class StreamOperatorTask implements StreamTask, InitableTask, WindowableTask, ClosableTask {
   private static final Logger LOG = LoggerFactory.getLogger(StreamOperatorTask.class);
 
   private final StreamApplication streamApplication;

samza-core/src/test/java/org/apache/samza/execution/TestExecutionPlanner.java

@@ -177,12 +177,12 @@ private StreamGraphImpl createStreamGraphWithJoinAndWindow() {
     OutputStream<KV<Object, Object>> output2 = streamGraph.getOutputStream("output2");
 
     messageStream1.map(m -> m)
-        .filter(m->true)
-        .window(Windows.keyedTumblingWindow(m -> m, Duration.ofMillis(8), mock(Serde.class), mock(Serde.class)));
+        .filter(m -> true)
+        .window(Windows.<KV<Object, Object>, Object>keyedTumblingWindow(m -> m, Duration.ofMillis(8), mock(Serde.class), mock(Serde.class)));
 
     messageStream2.map(m -> m)
-        .filter(m->true)
-        .window(Windows.keyedTumblingWindow(m -> m, Duration.ofMillis(16), mock(Serde.class), mock(Serde.class)));
+        .filter(m -> true)
+        .window(Windows.<KV<Object, Object>, Object>keyedTumblingWindow(m -> m, Duration.ofMillis(16), mock(Serde.class), mock(Serde.class)));
 
     messageStream1
         .join(messageStream2, mock(JoinFunction.class),

samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaStreamSpec.java

@@ -167,6 +167,15 @@ public StreamSpec copyWithPartitionCount(int partitionCount) {
     return new KafkaStreamSpec(getId(), getPhysicalName(), getSystemName(), partitionCount, getReplicationFactor(), getProperties());
   }
 
+  /**
+   * Make a copy of the spec with new properties
+   * @param properties properties of the Kafka stream
+   * @return new instance of {@link KafkaStreamSpec}
+   */
+  public KafkaStreamSpec copyWithProperties(Properties properties) {
+    return new KafkaStreamSpec(getId(), getPhysicalName(), getSystemName(), getPartitionCount(), getReplicationFactor(), properties);
+  }
+
   public int getReplicationFactor() {
     return replicationFactor;
   }

samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManager.scala

@@ -43,6 +43,9 @@ import scala.collection.mutable
  * keyed to that taskName. If there is no such message, no checkpoint data
  * exists.  The underlying log has a single partition into which all
  * checkpoints and TaskName to changelog partition mappings are written.
+ *
+ * This class is thread safe for writing but not for reading checkpoints.
+ * This is currently OK since checkpoints are only read on the main thread.
  */
 class KafkaCheckpointManager(
                               clientId: String,
@@ -64,14 +67,14 @@ class KafkaCheckpointManager(
                               checkpointTopicProperties: Properties = new Properties) extends CheckpointManager with Logging {
 
   var taskNames = Set[TaskName]()
-  @volatile var  systemProducer: SystemProducer = null
+  @volatile var systemProducer: SystemProducer = null
+  var systemConsumer: SystemConsumer = null
   var taskNamesToOffsets: Map[TaskName, Checkpoint] = null
   val systemAdmin = getSystemAdmin()
 
   val kafkaUtil: KafkaUtil = new KafkaUtil(retryBackoff, connectZk)
 
 
-
   KafkaCheckpointLogKey.setSystemStreamPartitionGrouperFactoryString(systemStreamPartitionGrouperFactoryString)
 
   info("Creating KafkaCheckpointManager with: clientId=%s, checkpointTopic=%s, systemName=%s" format(clientId, checkpointTopic, systemName))
@@ -181,73 +184,40 @@ class KafkaCheckpointManager(
    */
   private def readLog(shouldHandleEntry: (KafkaCheckpointLogKey) => Boolean,
                       handleEntry: (ByteBuffer, KafkaCheckpointLogKey) => Unit): Unit = {
-
-    val UNKNOWN_OFFSET = "-1"
-    var attempts = 10
-    val POLL_TIMEOUT = 1000L
+    info("Reading from checkpoint system:%s topic:%s" format(systemName, checkpointTopic))
 
     val ssp: SystemStreamPartition = new SystemStreamPartition(systemName, checkpointTopic, new Partition(0))
-    val systemConsumer = getSystemConsumer()
-    val partitionMetadata = getSSPMetadata(checkpointTopic, new Partition(0))
-    // offsets returned are strings
-    val newestOffset = if (partitionMetadata.getNewestOffset == null) UNKNOWN_OFFSET else partitionMetadata.getNewestOffset
-    val oldestOffset = partitionMetadata.getOldestOffset
-    systemConsumer.register(ssp, oldestOffset) // checkpoint stream should always be read from the beginning
-    systemConsumer.start()
 
-    var msgCount = 0
-    try {
-      val emptyEnvelopes = util.Collections.emptyMap[SystemStreamPartition, java.util.List[IncomingMessageEnvelope]]
-      // convert offsets to long
-      var currentOffset = UNKNOWN_OFFSET.toLong
-      val newestOffsetLong = newestOffset.toLong
-      val sspToPoll = Collections.singleton(ssp)
-      while (currentOffset < newestOffsetLong) {
-
-        val envelopes: java.util.Map[SystemStreamPartition, java.util.List[IncomingMessageEnvelope]] =
-        try {
-          systemConsumer.poll(sspToPoll, POLL_TIMEOUT)
-        } catch {
-          case e: Exception => {
-            // these exceptions are most likely intermediate
-            warn("Got %s exception while polling the consumer for checkpoints." format e)
-            if (attempts == 0) throw new SamzaException("Multiple attempts failed while reading the checkpoints. Giving up.", e)
-            attempts -= 1
-            emptyEnvelopes
-          }
-        }
+    if (systemConsumer == null) {
+      val partitionMetadata = getSSPMetadata(checkpointTopic, new Partition(0))
+      val oldestOffset = partitionMetadata.getOldestOffset
 
-        val messages: util.List[IncomingMessageEnvelope] = envelopes.get(ssp)
-        val messagesNum = if (messages != null) messages.size else 0
-        info("CheckpointMgr read %s envelopes (%s messages) from ssp %s. Current offset is %s, newest is %s"
-                     format (envelopes.size(), messagesNum, ssp, currentOffset, newestOffset))
-        if (envelopes.isEmpty || messagesNum <= 0) {
-          info("Got empty/null list of messages")
-        } else {
-          msgCount += messages.size()
-          // check the key
-          for (msg: IncomingMessageEnvelope <- messages) {
-            val key = msg.getKey.asInstanceOf[Array[Byte]]
-            currentOffset = msg.getOffset().toLong
-            if (key == null) {
-              throw new KafkaUtilException("While reading checkpoint (currentOffset=%s) stream encountered message without key."
-                                                   format currentOffset)
-            }
+      systemConsumer = getSystemConsumer()
+      systemConsumer.register(ssp, oldestOffset)
+      systemConsumer.start()
+    }
 
-            val checkpointKey = KafkaCheckpointLogKey.fromBytes(key)
+    val iterator =  new SystemStreamPartitionIterator(systemConsumer, ssp);
+    var msgCount = 0
+    while (iterator.hasNext) {
+      val msg = iterator.next
+      msgCount += 1
+
+      val offset = msg.getOffset
+      val key = msg.getKey.asInstanceOf[Array[Byte]]
+      if (key == null) {
+        throw new KafkaUtilException(
+          "While reading checkpoint (currentOffset=%s) stream encountered message without key." format offset)
+      }
 
-            if (!shouldHandleEntry(checkpointKey)) {
-              info("Skipping checkpoint log entry at offset %s with key %s." format(currentOffset, checkpointKey))
-            } else {
-              // handleEntry requires ByteBuffer
-              val checkpointPayload = ByteBuffer.wrap(msg.getMessage.asInstanceOf[Array[Byte]])
-              handleEntry(checkpointPayload, checkpointKey)
-            }
-          }
-        }
+      val checkpointKey = KafkaCheckpointLogKey.fromBytes(key)
+
+      if (!shouldHandleEntry(checkpointKey)) {
+        info("Skipping checkpoint log entry at offset %s with key %s." format(offset, checkpointKey))
+      } else {
+        val checkpointPayload = ByteBuffer.wrap(msg.getMessage.asInstanceOf[Array[Byte]])
+        handleEntry(checkpointPayload, checkpointKey)
       }
-    } finally {
-      systemConsumer.stop()
     }
     info("Done reading %s messages from checkpoint system:%s topic:%s" format(msgCount, systemName, checkpointTopic))
   }
@@ -282,12 +252,17 @@ class KafkaCheckpointManager(
 
 
   def stop = {
-    synchronized (
+    synchronized {
       if (systemProducer != null) {
         systemProducer.stop
         systemProducer = null
       }
-    )
+
+      if (systemConsumer != null) {
+        systemConsumer.stop
+        systemConsumer = null
+      }
+    }
 
   }
 

samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManagerFactory.scala

@@ -21,11 +21,13 @@ package org.apache.samza.checkpoint.kafka
 
 import java.util.Properties
 
+import com.google.common.collect.ImmutableMap
 import kafka.utils.ZkUtils
 import org.apache.samza.SamzaException
 import org.apache.samza.checkpoint.{CheckpointManager, CheckpointManagerFactory}
+import org.apache.samza.config.ApplicationConfig.ApplicationMode
 import org.apache.samza.config.JobConfig.Config2Job
-import org.apache.samza.config.{Config, KafkaConfig, SystemConfig}
+import org.apache.samza.config._
 import org.apache.samza.metrics.MetricsRegistry
 import org.apache.samza.system.SystemFactory
 import org.apache.samza.util.{ClientUtilTopicMetadataStore, KafkaUtil, Logging, Util, _}
@@ -38,21 +40,6 @@ object KafkaCheckpointManagerFactory {
     // on log compacted topics. Details in SAMZA-586.
     "compression.type" -> "none")
 
-  // Set the checkpoint topic configs to have a very small segment size and
-  // enable log compaction. This keeps job startup time small since there
-  // are fewer useless (overwritten) messages to read from the checkpoint
-  // topic.
-  def getCheckpointTopicProperties(config: Config) = {
-    val segmentBytes: Int = if (config == null) {
-      KafkaConfig.DEFAULT_CHECKPOINT_SEGMENT_BYTES
-    } else {
-      new KafkaConfig(config).getCheckpointSegmentBytes()
-    }
-    (new Properties /: Map(
-      "cleanup.policy" -> "compact",
-      "segment.bytes" -> String.valueOf(segmentBytes))) { case (props, (k, v)) => props.put(k, v); props }
-  }
-
   /**
    * Get the checkpoint system and system factory from the configuration
    * @param config
@@ -113,6 +100,6 @@ class KafkaCheckpointManagerFactory extends CheckpointManagerFactory with Loggin
       connectZk,
       config.getSystemStreamPartitionGrouperFactory,      // To find out the SSPGrouperFactory class so it can be included/verified in the key
       config.failOnCheckpointValidation,
-      checkpointTopicProperties = getCheckpointTopicProperties(config))
+      checkpointTopicProperties = new KafkaConfig(config).getCheckpointTopicProperties())
   }
 }