SAMZA-2041: add hdfs and kinesis descriptor

Author: Hai Lu <[email protected]>

Reviewers: Xinyu Liu <[email protected]>

Closes apache#857 from lhaiesp/master

Author: lhaiesp

docs/learn/documentation/versioned/connectors/hdfs.md

@@ -47,24 +47,22 @@ While streaming sources like Kafka are unbounded, files on HDFS have finite data
 
 #### Defining streams
 
-Samza uses the notion of a _system_ to describe any I/O source it interacts with. To consume from HDFS, you should create a new system that points to - `HdfsSystemFactory`. You can then associate multiple streams with this _system_. Each stream should have a _physical name_, which should be set to the name of the directory on HDFS.
-
-{% highlight jproperties %}
-systems.hdfs.samza.factory=org.apache.samza.system.hdfs.HdfsSystemFactory
-
-streams.hdfs-clickstream.samza.system=hdfs
-streams.hdfs-clickstream.samza.physical.name=hdfs:/data/clickstream/2016/09/11
+In Samza high level API, you can use `HdfsSystemDescriptor` to create a HDFS system. The stream name should be set to the name of the directory on HDFS.
 
+{% highlight java %}
+HdfsSystemDescriptor hsd = new HdfsSystemDescriptor("hdfs-clickstream");
+HdfsInputDescriptor hid = hsd.getInputDescriptor("/data/clickstream/2016/09/11");
 {% endhighlight %}
 
 The above example defines a stream called `hdfs-clickstream` that reads data from the `/data/clickstream/2016/09/11` directory. 
 
 #### Whitelists & Blacklists
 If you only want to consume from files that match a certain pattern, you can configure a whitelist. Likewise, you can also blacklist consuming from certain files. When both are specified, the _whitelist_ selects the files to be filtered and the _blacklist_ is later applied on its results. 
 
-{% highlight jproperties %}
-systems.hdfs.partitioner.defaultPartitioner.whitelist=.*avro
-systems.hdfs.partitioner.defaultPartitioner.blacklist=somefile.avro
+{% highlight java %}
+HdfsSystemDescriptor hsd = new HdfsSystemDescriptor("hdfs-clickstream")
+                                        .withConsumerWhiteList(".*avro")
+                                        .withConsumerBlackList("somefile.avro");
 {% endhighlight %}
 
 
@@ -74,34 +72,34 @@ systems.hdfs.partitioner.defaultPartitioner.blacklist=somefile.avro
 
 Samza allows writing your output results to HDFS in AVRO format. You can either use avro's GenericRecords or have Samza automatically infer the schema for your object using reflection. 
 
-{% highlight jproperties %}
-# set the SystemFactory implementation to instantiate HdfsSystemProducer aliased to 'hdfs'
-systems.hdfs.samza.factory=org.apache.samza.system.hdfs.HdfsSystemFactory
-systems.hdfs.producer.hdfs.writer.class=org.apache.samza.system.hdfs.writer.AvroDataFileHdfsWriter
+{% highlight java %}
+HdfsSystemDescriptor hsd = new HdfsSystemDescriptor("hdfs-clickstream")
+                                        .withWriterClassName(AvroDataFileHdfsWriter.class.getName());
 {% endhighlight %}
 
 
-If your output is non-avro, you can describe its format by implementing your own serializer.
-{% highlight jproperties %}
-systems.hdfs.producer.hdfs.writer.class=org.apache.samza.system.hdfs.writer.TextSequenceFileHdfsWriter
-serializers.registry.my-serde-name.class=MySerdeFactory
-systems.hdfs.samza.msg.serde=my-serde-name
+If your output is non-avro, use `TextSequenceFileHdfsWriter`.
+{% highlight java %}
+HdfsSystemDescriptor hsd = new HdfsSystemDescriptor("hdfs-clickstream")
+                                        .withWriterClassName(TextSequenceFileHdfsWriter.class.getName());
 {% endhighlight %}
 
 
 #### Output directory structure
 
 Samza allows you to control the base HDFS directory to write your output. You can also organize the output into sub-directories depending on the time your application ran, by configuring a date-formatter. 
-{% highlight jproperties %}
-systems.hdfs.producer.hdfs.base.output.dir=/user/me/analytics/clickstream_data
-systems.hdfs.producer.hdfs.bucketer.date.path.format=yyyy_MM_dd
+{% highlight java %}
+HdfsSystemDescriptor hsd = new HdfsSystemDescriptor("hdfs-clickstream")
+                                        .withOutputBaseDir("/user/me/analytics/clickstream_data")
+                                        .withDatePathFormat("yyyy_MM_dd");
 {% endhighlight %}
 
 You can configure the maximum size of each file or the maximum number of records per-file. Once either limits have been reached, Samza will create a new file.
 
-{% highlight jproperties %}
-systems.hdfs.producer.hdfs.write.batch.size.bytes=134217728
-systems.hdfs.producer.hdfs.write.batch.size.records=10000
+{% highlight java %}
+HdfsSystemDescriptor hsd = new HdfsSystemDescriptor("hdfs-clickstream")
+                                        .withWriteBatchSizeBytes(134217728)
+                                        .withWriteBatchSizeRecords(10000);
 {% endhighlight %}
 
 ### Security 

docs/learn/documentation/versioned/connectors/kinesis.md

@@ -36,22 +36,16 @@ wraps the Record into a [KinesisIncomingMessageEnvelope](https://github.com/apac
 
 #### Basic Configuration
 
-Here is the required configuration for consuming messages from Kinesis. 
-
-{% highlight jproperties %}
-// Define a Kinesis system factory with your identifier. eg: kinesis-system
-systems.kinesis-system.samza.factory=org.apache.samza.system.kinesis.KinesisSystemFactory
-
-// Kinesis consumer works with only AllSspToSingleTaskGrouperFactory
-job.systemstreampartition.grouper.factory=org.apache.samza.container.grouper.stream.AllSspToSingleTaskGrouperFactory
-
-// Define your streams
-task.inputs=kinesis-system.input0
-
-// Define required properties for your streams
-systems.kinesis-system.streams.input0.aws.region=YOUR-STREAM-REGION
-systems.kinesis-system.streams.input0.aws.accessKey=YOUR-ACCESS_KEY
-sensitive.systems.kinesis-system.streams.input0.aws.secretKey=YOUR-SECRET-KEY
+Here is the required configuration for consuming messages from Kinesis, through `KinesisSystemDescriptor` and `KinesisInputDescriptor`. 
+
+{% highlight java %}
+KinesisSystemDescriptor ksd = new KinesisSystemDescriptor("kinesis");
+    
+KinesisInputDescriptor<KV<String, byte[]>> kid = 
+    ksd.getInputDescriptor("STREAM-NAME", new NoOpSerde<byte[]>())
+          .withRegion("STREAM-REGION")
+          .withAccessKey("YOUR-ACCESS_KEY")
+          .withSecretKey("YOUR-SECRET-KEY");
 {% endhighlight %}
 
 ####Coordination
@@ -66,40 +60,57 @@ job.systemstreampartition.grouper.factory=org.apache.samza.container.grouper.str
 
 Each Kinesis stream in a given AWS [region](https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/Concepts.RegionsAndAvailabilityZones.html) can be accessed by providing an [access key](https://docs.aws.amazon.com/general/latest/gr/aws-sec-cred-types.html#access-keys-and-secret-access-keys). An Access key consists of two parts: an access key ID (for example, `AKIAIOSFODNN7EXAMPLE`) and a secret access key (for example, `wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY`) which you can use to send programmatic requests to AWS. 
 
-{% highlight jproperties %}
-systems.kinesis-system.streams.input0.aws.region=YOUR-STREAM-REGION
-systems.kinesis-system.streams.input0.aws.accessKey=YOUR-ACCESS_KEY
-sensitive.systems.kinesis-system.streams.input0.aws.secretKey=YOUR-SECRET-KEY
+{% highlight java %}
+KinesisInputDescriptor<KV<String, byte[]>> kid = 
+    ksd.getInputDescriptor("STREAM-NAME", new NoOpSerde<byte[]>())
+          .withRegion("STREAM-REGION")
+          .withAccessKey("YOUR-ACCESS_KEY")
+          .withSecretKey("YOUR-SECRET-KEY");
 {% endhighlight %}
 
 ### Advanced Configuration
 
 #### Kinesis Client Library Configs
 Samza Kinesis Connector uses the [Kinesis Client Library](https://docs.aws.amazon.com/streams/latest/dev/developing-consumers-with-kcl.html#kinesis-record-processor-overview-kcl)
 (KCL) to access the Kinesis data streams. You can set any [KCL Configuration](https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client-multilang/src/main/java/software/amazon/kinesis/coordinator/KinesisClientLibConfiguration.java)
-for a stream by configuring it with the **systems.system-name.streams.stream-name.aws.kcl.*** prefix.
+for a stream by configuring it through `KinesisInputDescriptor`.
 
-{% highlight jproperties %}
-systems.system-name.streams.stream-name.aws.kcl.CONFIG-PARAM=CONFIG-VALUE
+{% highlight java %}
+KinesisInputDescriptor<KV<String, byte[]>> kid = ...
+
+Map<String, String> kclConfig = new HashMap<>;
+kclConfig.put("CONFIG-PARAM", "CONFIG-VALUE");
+
+kid.withKCLConfig(kclConfig);
 {% endhighlight %}
 
 As an example, the below configuration is equivalent to invoking `kclClient#WithTableName(myTable)` on the KCL instance.
-{% highlight jproperties %}
-systems.system-name.streams.stream-name.aws.kcl.TableName=myTable
+{% highlight java %}
+KinesisInputDescriptor<KV<String, byte[]>> kid = ...
+
+Map<String, String> kclConfig = new HashMap<>;
+kclConfig.put("TableName", "myTable");
+
+kid.withKCLConfig(kclConfig);
 {% endhighlight %}
 
 #### AWS Client configs
 Samza allows you to specify any [AWS client configs](http://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/ClientConfiguration.html) to connect to your Kinesis instance.
-You can configure any [AWS client configuration](http://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/ClientConfiguration.html) with the `systems.your-system-name.aws.clientConfig.*` prefix.
+You can configure any [AWS client configuration](http://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/ClientConfiguration.html) through `KinesisSystemDescriptor`.
 
-{% highlight jproperties %}
-systems.system-name.aws.clientConfig.CONFIG-PARAM=CONFIG-VALUE
+{% highlight java %}
+Map<String, String> awsConfig = new HashMap<>;
+awsConfig.put("CONFIG-PARAM", "CONFIG-VALUE");
+
+KinesisSystemDescriptor sd = new KinesisSystemDescriptor(systemName)
+                                          .withAWSConfig(awsConfig);
 {% endhighlight %}
 
-As an example, to set the *proxy host* and *proxy port* to be used by the Kinesis Client:
-{% highlight jproperties %}
-systems.system-name.aws.clientConfig.ProxyHost=my-proxy-host.com
-systems.system-name.aws.clientConfig.ProxyPort=my-proxy-port
+Through `KinesisSystemDescriptor` you can also set the *proxy host* and *proxy port* to be used by the Kinesis Client:
+{% highlight java %}
+KinesisSystemDescriptor sd = new KinesisSystemDescriptor(systemName)
+                                          .withProxyHost("YOUR-PROXY-HOST")
+                                          .withProxyPort(YOUR-PROXY-PORT);
 {% endhighlight %}
 
 ### Resetting Offsets
@@ -109,14 +120,37 @@ These checkpoints are stored and managed by the KCL library internally. You can
 
 {% highlight jproperties %}
 // change the TableName to a unique name to reset checkpoints.
-systems.kinesis-system.streams.input0.aws.kcl.TableName=my-app-table-name
+systems.kinesis-system.streams.STREAM-NAME.aws.kcl.TableName=my-app-table-name
+{% endhighlight %}
+
+Or through `KinesisInputDescriptor`
+
+{% highlight java %}
+KinesisInputDescriptor<KV<String, byte[]>> kid = ...
+
+Map<String, String> kclConfig = new HashMap<>;
+kclConfig.put("TableName", "my-new-app-table-name");
+
+kid.withKCLConfig(kclConfig);
 {% endhighlight %}
 
+
 When you reset checkpoints, you can configure your job to start consuming from either the earliest or latest offset in the stream.  
 
 {% highlight jproperties %}
 // set the starting position to either TRIM_HORIZON (oldest) or LATEST (latest)
-systems.kinesis-system.streams.input0.aws.kcl.InitialPositionInStream=LATEST
+systems.kinesis-system.streams.STREAM-NAME.aws.kcl.InitialPositionInStream=LATEST
+{% endhighlight %}
+
+Or through `KinesisInputDescriptor`
+
+{% highlight java %}
+KinesisInputDescriptor<KV<String, byte[]>> kid = ...
+
+Map<String, String> kclConfig = new HashMap<>;
+kclConfig.put("InitialPositionInStream", "LATEST");
+
+kid.withKCLConfig(kclConfig);
 {% endhighlight %}
 
 Alternately, if you want to start from a particular offset in the Kinesis stream, you can login to the [AWS console](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/ConsoleDynamoDB.html) and edit the offsets in your DynamoDB Table.

samza-aws/src/main/java/org/apache/samza/system/kinesis/KinesisConfig.java

@@ -54,20 +54,20 @@
 public class KinesisConfig extends MapConfig {
   private static final Logger LOG = LoggerFactory.getLogger(KinesisConfig.class.getName());
 
-  private static final String CONFIG_SYSTEM_REGION = "systems.%s.aws.region";
-  private static final String CONFIG_STREAM_REGION = "systems.%s.streams.%s.aws.region";
+  public static final String CONFIG_SYSTEM_REGION = "systems.%s.aws.region";
+  public static final String CONFIG_STREAM_REGION = "systems.%s.streams.%s.aws.region";
 
-  private static final String CONFIG_STREAM_ACCESS_KEY = "systems.%s.streams.%s.aws.accessKey";
-  private static final String CONFIG_STREAM_SECRET_KEY = "sensitive.systems.%s.streams.%s.aws.secretKey";
+  public static final String CONFIG_STREAM_ACCESS_KEY = "systems.%s.streams.%s.aws.accessKey";
+  public static final String CONFIG_STREAM_SECRET_KEY = "sensitive.systems.%s.streams.%s.aws.secretKey";
 
-  private static final String CONFIG_AWS_CLIENT_CONFIG = "systems.%s.aws.clientConfig.";
-  private static final String CONFIG_PROXY_HOST = CONFIG_AWS_CLIENT_CONFIG + "ProxyHost";
-  private static final String DEFAULT_CONFIG_PROXY_HOST = "";
-  private static final String CONFIG_PROXY_PORT = CONFIG_AWS_CLIENT_CONFIG + "ProxyPort";
-  private static final int DEFAULT_CONFIG_PROXY_PORT = 0;
+  public static final String CONFIG_AWS_CLIENT_CONFIG = "systems.%s.aws.clientConfig.";
+  public static final String CONFIG_PROXY_HOST = CONFIG_AWS_CLIENT_CONFIG + "ProxyHost";
+  public static final String DEFAULT_CONFIG_PROXY_HOST = "";
+  public static final String CONFIG_PROXY_PORT = CONFIG_AWS_CLIENT_CONFIG + "ProxyPort";
+  public static final int DEFAULT_CONFIG_PROXY_PORT = 0;
 
-  private static final String CONFIG_SYSTEM_KINESIS_CLIENT_LIB_CONFIG = "systems.%s.aws.kcl.";
-  private static final String CONFIG_STREAM_KINESIS_CLIENT_LIB_CONFIG = "systems.%s.streams.%s.aws.kcl.";
+  public static final String CONFIG_SYSTEM_KINESIS_CLIENT_LIB_CONFIG = "systems.%s.aws.kcl.";
+  public static final String CONFIG_STREAM_KINESIS_CLIENT_LIB_CONFIG = "systems.%s.streams.%s.aws.kcl.";
 
   public KinesisConfig(Config config) {
     super(config);

samza-aws/src/main/java/org/apache/samza/system/kinesis/descriptors/KinesisInputDescriptor.java

@@ -0,0 +1,123 @@
+/*
+ * 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.system.kinesis.descriptors;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.samza.serializers.KVSerde;
+import org.apache.samza.serializers.NoOpSerde;
+import org.apache.samza.serializers.Serde;
+import org.apache.samza.system.descriptors.InputDescriptor;
+import org.apache.samza.system.descriptors.SystemDescriptor;
+import org.apache.samza.system.kinesis.KinesisConfig;
+
+
+/**
+ * A {@link KinesisInputDescriptor} can be used for specifying Samza and Kinesis specific properties of Kinesis
+ * input streams.
+ * <p>
+ * Use {@link KinesisSystemDescriptor#getInputDescriptor} to obtain an instance of this descriptor.
+ * <p>
+ * Stream properties provided in configuration override corresponding properties specified using a descriptor.
+ *
+ * @param <StreamMessageType> type of messages in this stream
+ */
+public class KinesisInputDescriptor<StreamMessageType>
+    extends InputDescriptor<StreamMessageType, KinesisInputDescriptor<StreamMessageType>> {
+  private Optional<String> accessKey = Optional.empty();
+  private Optional<String> secretKey = Optional.empty();
+  private Optional<String> region = Optional.empty();
+  private Map<String, String> kclConfig = Collections.emptyMap();
+
+
+  /**
+   * Constructs an {@link InputDescriptor} instance.
+   *
+   * @param streamId id of the stream
+   * @param valueSerde serde the values in the messages in the stream
+   * @param systemDescriptor system descriptor this stream descriptor was obtained from
+   */
+  <T> KinesisInputDescriptor(String streamId, Serde<T> valueSerde, SystemDescriptor systemDescriptor) {
+    super(streamId, KVSerde.of(new NoOpSerde<>(), valueSerde), systemDescriptor, null);
+  }
+
+  /**
+   * Kinesis region for the system stream.
+   * @param region Kinesis region
+   * @return this input descriptor
+   */
+  public KinesisInputDescriptor<StreamMessageType> withRegion(String region) {
+    this.region = Optional.of(StringUtils.stripToNull(region));
+    return this;
+  }
+
+  /**
+   * Kinesis access key name for the system stream.
+   * @param accessKey Kinesis access key name
+   * @return this input descriptor
+   */
+  public KinesisInputDescriptor<StreamMessageType> withAccessKey(String accessKey) {
+    this.accessKey = Optional.of(StringUtils.stripToNull(accessKey));
+    return this;
+  }
+
+  /**
+   * Kinesis secret key name for the system stream.
+   * @param secretKey Kinesis secret key
+   * @return this input descriptor
+   */
+  public KinesisInputDescriptor<StreamMessageType> withSecretKey(String secretKey) {
+    this.secretKey = Optional.of(StringUtils.stripToNull(secretKey));
+    return this;
+  }
+
+  /**
+   * KCL (Kinesis Client Library) config for the system stream. This is not required by default.
+   * @param kclConfig A map of specified KCL configs
+   * @return this input descriptor
+   */
+  public KinesisInputDescriptor<StreamMessageType> withKCLConfig(Map<String, String> kclConfig) {
+    this.kclConfig = kclConfig;
+    return this;
+  }
+
+  @Override
+  public Map<String, String> toConfig() {
+    Map<String, String> config = new HashMap<>(super.toConfig());
+
+    String systemName = getSystemName();
+    String streamId = getStreamId();
+    String clientConfigPrefix =
+        String.format(KinesisConfig.CONFIG_STREAM_KINESIS_CLIENT_LIB_CONFIG, systemName, streamId);
+
+    this.region.ifPresent(
+        val -> config.put(String.format(KinesisConfig.CONFIG_STREAM_REGION, systemName, streamId), val));
+    this.accessKey.ifPresent(
+        val -> config.put(String.format(KinesisConfig.CONFIG_STREAM_ACCESS_KEY, systemName, streamId), val));
+    this.secretKey.ifPresent(
+        val -> config.put(String.format(KinesisConfig.CONFIG_STREAM_SECRET_KEY, systemName, streamId), val));
+    this.kclConfig.forEach((k, v) -> config.put(clientConfigPrefix + k, v));
+
+    return config;
+  }
+}