You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by br...@apache.org on 2017/12/26 07:50:51 UTC
spark git commit: [SPARK-20168][DSTREAM] Add changes to use kinesis
fetches from specific timestamp
Repository: spark
Updated Branches:
refs/heads/master be03d3ad7 -> 0e6833006
[SPARK-20168][DSTREAM] Add changes to use kinesis fetches from specific timestamp
## What changes were proposed in this pull request?
Kinesis client can resume from a specified timestamp while creating a stream. We should have option to pass a timestamp in config to allow kinesis to resume from the given timestamp.
The patch introduces a new `KinesisInitialPositionInStream` that takes the `InitialPositionInStream` with the `timestamp` information that can be used to resume kinesis fetches from the provided timestamp.
## How was this patch tested?
Unit Tests
cc : budde brkyvz
Author: Yash Sharma <ys...@atlassian.com>
Closes #18029 from yssharma/ysharma/kcl_resume.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/0e683300
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/0e683300
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/0e683300
Branch: refs/heads/master
Commit: 0e6833006d28df426eb132bb8fc82917b8e2aedd
Parents: be03d3a
Author: Yash Sharma <ys...@atlassian.com>
Authored: Tue Dec 26 09:50:39 2017 +0200
Committer: Burak Yavuz <br...@gmail.com>
Committed: Tue Dec 26 09:50:39 2017 +0200
----------------------------------------------------------------------
.../kinesis/KinesisInitialPositions.java | 91 ++++++++++++++++++++
.../streaming/KinesisWordCountASL.scala | 5 +-
.../streaming/kinesis/KinesisInputDStream.scala | 31 +++++--
.../streaming/kinesis/KinesisReceiver.scala | 45 ++++++----
.../spark/streaming/kinesis/KinesisUtils.scala | 15 ++--
.../JavaKinesisInputDStreamBuilderSuite.java | 47 ++++++++--
.../KinesisInputDStreamBuilderSuite.scala | 68 +++++++++++++--
.../streaming/kinesis/KinesisStreamSuite.scala | 11 +--
8 files changed, 264 insertions(+), 49 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/0e683300/external/kinesis-asl/src/main/java/org/apache/spark/streaming/kinesis/KinesisInitialPositions.java
----------------------------------------------------------------------
diff --git a/external/kinesis-asl/src/main/java/org/apache/spark/streaming/kinesis/KinesisInitialPositions.java b/external/kinesis-asl/src/main/java/org/apache/spark/streaming/kinesis/KinesisInitialPositions.java
new file mode 100644
index 0000000..206e1e4
--- /dev/null
+++ b/external/kinesis-asl/src/main/java/org/apache/spark/streaming/kinesis/KinesisInitialPositions.java
@@ -0,0 +1,91 @@
+/*
+ * 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.spark.streaming.kinesis;
+
+import com.amazonaws.services.kinesis.clientlibrary.lib.worker.InitialPositionInStream;
+
+import java.io.Serializable;
+import java.util.Date;
+
+/**
+ * A java wrapper for exposing [[InitialPositionInStream]]
+ * to the corresponding Kinesis readers.
+ */
+interface KinesisInitialPosition {
+ InitialPositionInStream getPosition();
+}
+
+public class KinesisInitialPositions {
+ public static class Latest implements KinesisInitialPosition, Serializable {
+ public Latest() {}
+
+ @Override
+ public InitialPositionInStream getPosition() {
+ return InitialPositionInStream.LATEST;
+ }
+ }
+
+ public static class TrimHorizon implements KinesisInitialPosition, Serializable {
+ public TrimHorizon() {}
+
+ @Override
+ public InitialPositionInStream getPosition() {
+ return InitialPositionInStream.TRIM_HORIZON;
+ }
+ }
+
+ public static class AtTimestamp implements KinesisInitialPosition, Serializable {
+ private Date timestamp;
+
+ public AtTimestamp(Date timestamp) {
+ this.timestamp = timestamp;
+ }
+
+ @Override
+ public InitialPositionInStream getPosition() {
+ return InitialPositionInStream.AT_TIMESTAMP;
+ }
+
+ public Date getTimestamp() {
+ return timestamp;
+ }
+ }
+
+
+ /**
+ * Returns instance of [[KinesisInitialPosition]] based on the passed [[InitialPositionInStream]].
+ * This method is used in KinesisUtils for translating the InitialPositionInStream
+ * to InitialPosition. This function would be removed when we deprecate the KinesisUtils.
+ *
+ * @return [[InitialPosition]]
+ */
+ public static KinesisInitialPosition fromKinesisInitialPosition(
+ InitialPositionInStream initialPositionInStream) throws UnsupportedOperationException {
+ if (initialPositionInStream == InitialPositionInStream.LATEST) {
+ return new Latest();
+ } else if (initialPositionInStream == InitialPositionInStream.TRIM_HORIZON) {
+ return new TrimHorizon();
+ } else {
+ // InitialPositionInStream.AT_TIMESTAMP is not supported.
+ // Use InitialPosition.atTimestamp(timestamp) instead.
+ throw new UnsupportedOperationException(
+ "Only InitialPositionInStream.LATEST and InitialPositionInStream.TRIM_HORIZON " +
+ "supported in initialPositionInStream(). Please use the initialPosition() from " +
+ "builder API in KinesisInputDStream for using InitialPositionInStream.AT_TIMESTAMP");
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/spark/blob/0e683300/external/kinesis-asl/src/main/scala/org/apache/spark/examples/streaming/KinesisWordCountASL.scala
----------------------------------------------------------------------
diff --git a/external/kinesis-asl/src/main/scala/org/apache/spark/examples/streaming/KinesisWordCountASL.scala b/external/kinesis-asl/src/main/scala/org/apache/spark/examples/streaming/KinesisWordCountASL.scala
index cde2c4b..fcb790e 100644
--- a/external/kinesis-asl/src/main/scala/org/apache/spark/examples/streaming/KinesisWordCountASL.scala
+++ b/external/kinesis-asl/src/main/scala/org/apache/spark/examples/streaming/KinesisWordCountASL.scala
@@ -24,7 +24,6 @@ import scala.util.Random
import com.amazonaws.auth.DefaultAWSCredentialsProviderChain
import com.amazonaws.services.kinesis.AmazonKinesisClient
-import com.amazonaws.services.kinesis.clientlibrary.lib.worker.InitialPositionInStream
import com.amazonaws.services.kinesis.model.PutRecordRequest
import org.apache.log4j.{Level, Logger}
@@ -33,9 +32,9 @@ import org.apache.spark.internal.Logging
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.{Milliseconds, StreamingContext}
import org.apache.spark.streaming.dstream.DStream.toPairDStreamFunctions
+import org.apache.spark.streaming.kinesis.KinesisInitialPositions.Latest
import org.apache.spark.streaming.kinesis.KinesisInputDStream
-
/**
* Consumes messages from a Amazon Kinesis streams and does wordcount.
*
@@ -139,7 +138,7 @@ object KinesisWordCountASL extends Logging {
.streamName(streamName)
.endpointUrl(endpointUrl)
.regionName(regionName)
- .initialPositionInStream(InitialPositionInStream.LATEST)
+ .initialPosition(new Latest())
.checkpointAppName(appName)
.checkpointInterval(kinesisCheckpointInterval)
.storageLevel(StorageLevel.MEMORY_AND_DISK_2)
http://git-wip-us.apache.org/repos/asf/spark/blob/0e683300/external/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisInputDStream.scala
----------------------------------------------------------------------
diff --git a/external/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisInputDStream.scala b/external/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisInputDStream.scala
index f61e398..1ffec01 100644
--- a/external/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisInputDStream.scala
+++ b/external/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisInputDStream.scala
@@ -28,6 +28,7 @@ import org.apache.spark.storage.{BlockId, StorageLevel}
import org.apache.spark.streaming.{Duration, StreamingContext, Time}
import org.apache.spark.streaming.api.java.JavaStreamingContext
import org.apache.spark.streaming.dstream.ReceiverInputDStream
+import org.apache.spark.streaming.kinesis.KinesisInitialPositions.Latest
import org.apache.spark.streaming.receiver.Receiver
import org.apache.spark.streaming.scheduler.ReceivedBlockInfo
@@ -36,7 +37,7 @@ private[kinesis] class KinesisInputDStream[T: ClassTag](
val streamName: String,
val endpointUrl: String,
val regionName: String,
- val initialPositionInStream: InitialPositionInStream,
+ val initialPosition: KinesisInitialPosition,
val checkpointAppName: String,
val checkpointInterval: Duration,
val _storageLevel: StorageLevel,
@@ -77,7 +78,7 @@ private[kinesis] class KinesisInputDStream[T: ClassTag](
}
override def getReceiver(): Receiver[T] = {
- new KinesisReceiver(streamName, endpointUrl, regionName, initialPositionInStream,
+ new KinesisReceiver(streamName, endpointUrl, regionName, initialPosition,
checkpointAppName, checkpointInterval, _storageLevel, messageHandler,
kinesisCreds, dynamoDBCreds, cloudWatchCreds)
}
@@ -100,7 +101,7 @@ object KinesisInputDStream {
// Params with defaults
private var endpointUrl: Option[String] = None
private var regionName: Option[String] = None
- private var initialPositionInStream: Option[InitialPositionInStream] = None
+ private var initialPosition: Option[KinesisInitialPosition] = None
private var checkpointInterval: Option[Duration] = None
private var storageLevel: Option[StorageLevel] = None
private var kinesisCredsProvider: Option[SparkAWSCredentials] = None
@@ -182,14 +183,30 @@ object KinesisInputDStream {
/**
* Sets the initial position data is read from in the Kinesis stream. Defaults to
+ * [[KinesisInitialPositions.Latest]] if no custom value is specified.
+ *
+ * @param initialPosition [[KinesisInitialPosition]] value specifying where Spark Streaming
+ * will start reading records in the Kinesis stream from
+ * @return Reference to this [[KinesisInputDStream.Builder]]
+ */
+ def initialPosition(initialPosition: KinesisInitialPosition): Builder = {
+ this.initialPosition = Option(initialPosition)
+ this
+ }
+
+ /**
+ * Sets the initial position data is read from in the Kinesis stream. Defaults to
* [[InitialPositionInStream.LATEST]] if no custom value is specified.
+ * This function would be removed when we deprecate the KinesisUtils.
*
* @param initialPosition InitialPositionInStream value specifying where Spark Streaming
* will start reading records in the Kinesis stream from
* @return Reference to this [[KinesisInputDStream.Builder]]
*/
+ @deprecated("use initialPosition(initialPosition: KinesisInitialPosition)", "2.3.0")
def initialPositionInStream(initialPosition: InitialPositionInStream): Builder = {
- initialPositionInStream = Option(initialPosition)
+ this.initialPosition = Option(
+ KinesisInitialPositions.fromKinesisInitialPosition(initialPosition))
this
}
@@ -266,7 +283,7 @@ object KinesisInputDStream {
getRequiredParam(streamName, "streamName"),
endpointUrl.getOrElse(DEFAULT_KINESIS_ENDPOINT_URL),
regionName.getOrElse(DEFAULT_KINESIS_REGION_NAME),
- initialPositionInStream.getOrElse(DEFAULT_INITIAL_POSITION_IN_STREAM),
+ initialPosition.getOrElse(DEFAULT_INITIAL_POSITION),
getRequiredParam(checkpointAppName, "checkpointAppName"),
checkpointInterval.getOrElse(ssc.graph.batchDuration),
storageLevel.getOrElse(DEFAULT_STORAGE_LEVEL),
@@ -293,7 +310,6 @@ object KinesisInputDStream {
* Creates a [[KinesisInputDStream.Builder]] for constructing [[KinesisInputDStream]] instances.
*
* @since 2.2.0
- *
* @return [[KinesisInputDStream.Builder]] instance
*/
def builder: Builder = new Builder
@@ -309,7 +325,6 @@ object KinesisInputDStream {
private[kinesis] val DEFAULT_KINESIS_ENDPOINT_URL: String =
"https://kinesis.us-east-1.amazonaws.com"
private[kinesis] val DEFAULT_KINESIS_REGION_NAME: String = "us-east-1"
- private[kinesis] val DEFAULT_INITIAL_POSITION_IN_STREAM: InitialPositionInStream =
- InitialPositionInStream.LATEST
+ private[kinesis] val DEFAULT_INITIAL_POSITION: KinesisInitialPosition = new Latest()
private[kinesis] val DEFAULT_STORAGE_LEVEL: StorageLevel = StorageLevel.MEMORY_AND_DISK_2
}
http://git-wip-us.apache.org/repos/asf/spark/blob/0e683300/external/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisReceiver.scala
----------------------------------------------------------------------
diff --git a/external/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisReceiver.scala b/external/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisReceiver.scala
index 1026d0f..fa0de62 100644
--- a/external/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisReceiver.scala
+++ b/external/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisReceiver.scala
@@ -24,12 +24,13 @@ import scala.collection.mutable
import scala.util.control.NonFatal
import com.amazonaws.services.kinesis.clientlibrary.interfaces.{IRecordProcessor, IRecordProcessorCheckpointer, IRecordProcessorFactory}
-import com.amazonaws.services.kinesis.clientlibrary.lib.worker.{InitialPositionInStream, KinesisClientLibConfiguration, Worker}
+import com.amazonaws.services.kinesis.clientlibrary.lib.worker.{KinesisClientLibConfiguration, Worker}
import com.amazonaws.services.kinesis.model.Record
import org.apache.spark.internal.Logging
import org.apache.spark.storage.{StorageLevel, StreamBlockId}
import org.apache.spark.streaming.Duration
+import org.apache.spark.streaming.kinesis.KinesisInitialPositions.AtTimestamp
import org.apache.spark.streaming.receiver.{BlockGenerator, BlockGeneratorListener, Receiver}
import org.apache.spark.util.Utils
@@ -56,12 +57,13 @@ import org.apache.spark.util.Utils
* @param endpointUrl Url of Kinesis service (e.g., https://kinesis.us-east-1.amazonaws.com)
* @param regionName Region name used by the Kinesis Client Library for
* DynamoDB (lease coordination and checkpointing) and CloudWatch (metrics)
- * @param initialPositionInStream In the absence of Kinesis checkpoint info, this is the
- * worker's initial starting position in the stream.
- * The values are either the beginning of the stream
- * per Kinesis' limit of 24 hours
- * (InitialPositionInStream.TRIM_HORIZON) or
- * the tip of the stream (InitialPositionInStream.LATEST).
+ * @param initialPosition Instance of [[KinesisInitialPosition]]
+ * In the absence of Kinesis checkpoint info, this is the
+ * worker's initial starting position in the stream.
+ * The values are either the beginning of the stream
+ * per Kinesis' limit of 24 hours
+ * ([[KinesisInitialPositions.TrimHorizon]]) or
+ * the tip of the stream ([[KinesisInitialPositions.Latest]]).
* @param checkpointAppName Kinesis application name. Kinesis Apps are mapped to Kinesis Streams
* by the Kinesis Client Library. If you change the App name or Stream name,
* the KCL will throw errors. This usually requires deleting the backing
@@ -83,7 +85,7 @@ private[kinesis] class KinesisReceiver[T](
val streamName: String,
endpointUrl: String,
regionName: String,
- initialPositionInStream: InitialPositionInStream,
+ initialPosition: KinesisInitialPosition,
checkpointAppName: String,
checkpointInterval: Duration,
storageLevel: StorageLevel,
@@ -148,18 +150,29 @@ private[kinesis] class KinesisReceiver[T](
kinesisCheckpointer = new KinesisCheckpointer(receiver, checkpointInterval, workerId)
val kinesisProvider = kinesisCreds.provider
- val kinesisClientLibConfiguration = new KinesisClientLibConfiguration(
- checkpointAppName,
- streamName,
- kinesisProvider,
- dynamoDBCreds.map(_.provider).getOrElse(kinesisProvider),
- cloudWatchCreds.map(_.provider).getOrElse(kinesisProvider),
- workerId)
+
+ val kinesisClientLibConfiguration = {
+ val baseClientLibConfiguration = new KinesisClientLibConfiguration(
+ checkpointAppName,
+ streamName,
+ kinesisProvider,
+ dynamoDBCreds.map(_.provider).getOrElse(kinesisProvider),
+ cloudWatchCreds.map(_.provider).getOrElse(kinesisProvider),
+ workerId)
.withKinesisEndpoint(endpointUrl)
- .withInitialPositionInStream(initialPositionInStream)
+ .withInitialPositionInStream(initialPosition.getPosition)
.withTaskBackoffTimeMillis(500)
.withRegionName(regionName)
+ // Update the Kinesis client lib config with timestamp
+ // if InitialPositionInStream.AT_TIMESTAMP is passed
+ initialPosition match {
+ case ts: AtTimestamp =>
+ baseClientLibConfiguration.withTimestampAtInitialPositionInStream(ts.getTimestamp)
+ case _ => baseClientLibConfiguration
+ }
+ }
+
/*
* RecordProcessorFactory creates impls of IRecordProcessor.
* IRecordProcessor adapts the KCL to our Spark KinesisReceiver via the
http://git-wip-us.apache.org/repos/asf/spark/blob/0e683300/external/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisUtils.scala
----------------------------------------------------------------------
diff --git a/external/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisUtils.scala b/external/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisUtils.scala
index 1298463..2500460 100644
--- a/external/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisUtils.scala
+++ b/external/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisUtils.scala
@@ -73,7 +73,8 @@ object KinesisUtils {
// Setting scope to override receiver stream's scope of "receiver stream"
ssc.withNamedScope("kinesis stream") {
new KinesisInputDStream[T](ssc, streamName, endpointUrl, validateRegion(regionName),
- initialPositionInStream, kinesisAppName, checkpointInterval, storageLevel,
+ KinesisInitialPositions.fromKinesisInitialPosition(initialPositionInStream),
+ kinesisAppName, checkpointInterval, storageLevel,
cleanedHandler, DefaultCredentials, None, None)
}
}
@@ -129,7 +130,8 @@ object KinesisUtils {
awsAccessKeyId = awsAccessKeyId,
awsSecretKey = awsSecretKey)
new KinesisInputDStream[T](ssc, streamName, endpointUrl, validateRegion(regionName),
- initialPositionInStream, kinesisAppName, checkpointInterval, storageLevel,
+ KinesisInitialPositions.fromKinesisInitialPosition(initialPositionInStream),
+ kinesisAppName, checkpointInterval, storageLevel,
cleanedHandler, kinesisCredsProvider, None, None)
}
}
@@ -198,7 +200,8 @@ object KinesisUtils {
awsAccessKeyId = awsAccessKeyId,
awsSecretKey = awsSecretKey))
new KinesisInputDStream[T](ssc, streamName, endpointUrl, validateRegion(regionName),
- initialPositionInStream, kinesisAppName, checkpointInterval, storageLevel,
+ KinesisInitialPositions.fromKinesisInitialPosition(initialPositionInStream),
+ kinesisAppName, checkpointInterval, storageLevel,
cleanedHandler, kinesisCredsProvider, None, None)
}
}
@@ -243,7 +246,8 @@ object KinesisUtils {
// Setting scope to override receiver stream's scope of "receiver stream"
ssc.withNamedScope("kinesis stream") {
new KinesisInputDStream[Array[Byte]](ssc, streamName, endpointUrl, validateRegion(regionName),
- initialPositionInStream, kinesisAppName, checkpointInterval, storageLevel,
+ KinesisInitialPositions.fromKinesisInitialPosition(initialPositionInStream),
+ kinesisAppName, checkpointInterval, storageLevel,
KinesisInputDStream.defaultMessageHandler, DefaultCredentials, None, None)
}
}
@@ -293,7 +297,8 @@ object KinesisUtils {
awsAccessKeyId = awsAccessKeyId,
awsSecretKey = awsSecretKey)
new KinesisInputDStream[Array[Byte]](ssc, streamName, endpointUrl, validateRegion(regionName),
- initialPositionInStream, kinesisAppName, checkpointInterval, storageLevel,
+ KinesisInitialPositions.fromKinesisInitialPosition(initialPositionInStream),
+ kinesisAppName, checkpointInterval, storageLevel,
KinesisInputDStream.defaultMessageHandler, kinesisCredsProvider, None, None)
}
}
http://git-wip-us.apache.org/repos/asf/spark/blob/0e683300/external/kinesis-asl/src/test/java/org/apache/spark/streaming/kinesis/JavaKinesisInputDStreamBuilderSuite.java
----------------------------------------------------------------------
diff --git a/external/kinesis-asl/src/test/java/org/apache/spark/streaming/kinesis/JavaKinesisInputDStreamBuilderSuite.java b/external/kinesis-asl/src/test/java/org/apache/spark/streaming/kinesis/JavaKinesisInputDStreamBuilderSuite.java
index be6d549..03becd7 100644
--- a/external/kinesis-asl/src/test/java/org/apache/spark/streaming/kinesis/JavaKinesisInputDStreamBuilderSuite.java
+++ b/external/kinesis-asl/src/test/java/org/apache/spark/streaming/kinesis/JavaKinesisInputDStreamBuilderSuite.java
@@ -17,14 +17,13 @@
package org.apache.spark.streaming.kinesis;
-import org.junit.Test;
-
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.InitialPositionInStream;
-
+import org.apache.spark.streaming.kinesis.KinesisInitialPositions.TrimHorizon;
import org.apache.spark.storage.StorageLevel;
import org.apache.spark.streaming.Duration;
-import org.apache.spark.streaming.Seconds;
import org.apache.spark.streaming.LocalJavaStreamingContext;
+import org.apache.spark.streaming.Seconds;
+import org.junit.Test;
public class JavaKinesisInputDStreamBuilderSuite extends LocalJavaStreamingContext {
/**
@@ -35,7 +34,41 @@ public class JavaKinesisInputDStreamBuilderSuite extends LocalJavaStreamingConte
String streamName = "a-very-nice-stream-name";
String endpointUrl = "https://kinesis.us-west-2.amazonaws.com";
String region = "us-west-2";
- InitialPositionInStream initialPosition = InitialPositionInStream.TRIM_HORIZON;
+ KinesisInitialPosition initialPosition = new TrimHorizon();
+ String appName = "a-very-nice-kinesis-app";
+ Duration checkpointInterval = Seconds.apply(30);
+ StorageLevel storageLevel = StorageLevel.MEMORY_ONLY();
+
+ KinesisInputDStream<byte[]> kinesisDStream = KinesisInputDStream.builder()
+ .streamingContext(ssc)
+ .streamName(streamName)
+ .endpointUrl(endpointUrl)
+ .regionName(region)
+ .initialPosition(initialPosition)
+ .checkpointAppName(appName)
+ .checkpointInterval(checkpointInterval)
+ .storageLevel(storageLevel)
+ .build();
+ assert(kinesisDStream.streamName() == streamName);
+ assert(kinesisDStream.endpointUrl() == endpointUrl);
+ assert(kinesisDStream.regionName() == region);
+ assert(kinesisDStream.initialPosition().getPosition() == initialPosition.getPosition());
+ assert(kinesisDStream.checkpointAppName() == appName);
+ assert(kinesisDStream.checkpointInterval() == checkpointInterval);
+ assert(kinesisDStream._storageLevel() == storageLevel);
+ ssc.stop();
+ }
+
+ /**
+ * Test to ensure that the old API for InitialPositionInStream
+ * is supported in KinesisDStream.Builder.
+ * This test would be removed when we deprecate the KinesisUtils.
+ */
+ @Test
+ public void testJavaKinesisDStreamBuilderOldApi() {
+ String streamName = "a-very-nice-stream-name";
+ String endpointUrl = "https://kinesis.us-west-2.amazonaws.com";
+ String region = "us-west-2";
String appName = "a-very-nice-kinesis-app";
Duration checkpointInterval = Seconds.apply(30);
StorageLevel storageLevel = StorageLevel.MEMORY_ONLY();
@@ -45,7 +78,7 @@ public class JavaKinesisInputDStreamBuilderSuite extends LocalJavaStreamingConte
.streamName(streamName)
.endpointUrl(endpointUrl)
.regionName(region)
- .initialPositionInStream(initialPosition)
+ .initialPositionInStream(InitialPositionInStream.LATEST)
.checkpointAppName(appName)
.checkpointInterval(checkpointInterval)
.storageLevel(storageLevel)
@@ -53,7 +86,7 @@ public class JavaKinesisInputDStreamBuilderSuite extends LocalJavaStreamingConte
assert(kinesisDStream.streamName() == streamName);
assert(kinesisDStream.endpointUrl() == endpointUrl);
assert(kinesisDStream.regionName() == region);
- assert(kinesisDStream.initialPositionInStream() == initialPosition);
+ assert(kinesisDStream.initialPosition().getPosition() == InitialPositionInStream.LATEST);
assert(kinesisDStream.checkpointAppName() == appName);
assert(kinesisDStream.checkpointInterval() == checkpointInterval);
assert(kinesisDStream._storageLevel() == storageLevel);
http://git-wip-us.apache.org/repos/asf/spark/blob/0e683300/external/kinesis-asl/src/test/scala/org/apache/spark/streaming/kinesis/KinesisInputDStreamBuilderSuite.scala
----------------------------------------------------------------------
diff --git a/external/kinesis-asl/src/test/scala/org/apache/spark/streaming/kinesis/KinesisInputDStreamBuilderSuite.scala b/external/kinesis-asl/src/test/scala/org/apache/spark/streaming/kinesis/KinesisInputDStreamBuilderSuite.scala
index afa1a7f..e0e2684 100644
--- a/external/kinesis-asl/src/test/scala/org/apache/spark/streaming/kinesis/KinesisInputDStreamBuilderSuite.scala
+++ b/external/kinesis-asl/src/test/scala/org/apache/spark/streaming/kinesis/KinesisInputDStreamBuilderSuite.scala
@@ -17,12 +17,15 @@
package org.apache.spark.streaming.kinesis
+import java.util.Calendar
+
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.InitialPositionInStream
import org.scalatest.BeforeAndAfterEach
import org.scalatest.mockito.MockitoSugar
import org.apache.spark.storage.StorageLevel
-import org.apache.spark.streaming.{Seconds, StreamingContext, TestSuiteBase}
+import org.apache.spark.streaming.{Duration, Seconds, StreamingContext, TestSuiteBase}
+import org.apache.spark.streaming.kinesis.KinesisInitialPositions.{AtTimestamp, TrimHorizon}
class KinesisInputDStreamBuilderSuite extends TestSuiteBase with BeforeAndAfterEach
with MockitoSugar {
@@ -69,7 +72,7 @@ class KinesisInputDStreamBuilderSuite extends TestSuiteBase with BeforeAndAfterE
val dstream = builder.build()
assert(dstream.endpointUrl == DEFAULT_KINESIS_ENDPOINT_URL)
assert(dstream.regionName == DEFAULT_KINESIS_REGION_NAME)
- assert(dstream.initialPositionInStream == DEFAULT_INITIAL_POSITION_IN_STREAM)
+ assert(dstream.initialPosition == DEFAULT_INITIAL_POSITION)
assert(dstream.checkpointInterval == batchDuration)
assert(dstream._storageLevel == DEFAULT_STORAGE_LEVEL)
assert(dstream.kinesisCreds == DefaultCredentials)
@@ -80,7 +83,7 @@ class KinesisInputDStreamBuilderSuite extends TestSuiteBase with BeforeAndAfterE
test("should propagate custom non-auth values to KinesisInputDStream") {
val customEndpointUrl = "https://kinesis.us-west-2.amazonaws.com"
val customRegion = "us-west-2"
- val customInitialPosition = InitialPositionInStream.TRIM_HORIZON
+ val customInitialPosition = new TrimHorizon()
val customAppName = "a-very-nice-kinesis-app"
val customCheckpointInterval = Seconds(30)
val customStorageLevel = StorageLevel.MEMORY_ONLY
@@ -91,7 +94,7 @@ class KinesisInputDStreamBuilderSuite extends TestSuiteBase with BeforeAndAfterE
val dstream = builder
.endpointUrl(customEndpointUrl)
.regionName(customRegion)
- .initialPositionInStream(customInitialPosition)
+ .initialPosition(customInitialPosition)
.checkpointAppName(customAppName)
.checkpointInterval(customCheckpointInterval)
.storageLevel(customStorageLevel)
@@ -101,12 +104,67 @@ class KinesisInputDStreamBuilderSuite extends TestSuiteBase with BeforeAndAfterE
.build()
assert(dstream.endpointUrl == customEndpointUrl)
assert(dstream.regionName == customRegion)
- assert(dstream.initialPositionInStream == customInitialPosition)
+ assert(dstream.initialPosition == customInitialPosition)
assert(dstream.checkpointAppName == customAppName)
assert(dstream.checkpointInterval == customCheckpointInterval)
assert(dstream._storageLevel == customStorageLevel)
assert(dstream.kinesisCreds == customKinesisCreds)
assert(dstream.dynamoDBCreds == Option(customDynamoDBCreds))
assert(dstream.cloudWatchCreds == Option(customCloudWatchCreds))
+
+ // Testing with AtTimestamp
+ val cal = Calendar.getInstance()
+ cal.add(Calendar.DATE, -1)
+ val timestamp = cal.getTime()
+ val initialPositionAtTimestamp = new AtTimestamp(timestamp)
+
+ val dstreamAtTimestamp = builder
+ .endpointUrl(customEndpointUrl)
+ .regionName(customRegion)
+ .initialPosition(initialPositionAtTimestamp)
+ .checkpointAppName(customAppName)
+ .checkpointInterval(customCheckpointInterval)
+ .storageLevel(customStorageLevel)
+ .kinesisCredentials(customKinesisCreds)
+ .dynamoDBCredentials(customDynamoDBCreds)
+ .cloudWatchCredentials(customCloudWatchCreds)
+ .build()
+ assert(dstreamAtTimestamp.endpointUrl == customEndpointUrl)
+ assert(dstreamAtTimestamp.regionName == customRegion)
+ assert(dstreamAtTimestamp.initialPosition.getPosition
+ == initialPositionAtTimestamp.getPosition)
+ assert(
+ dstreamAtTimestamp.initialPosition.asInstanceOf[AtTimestamp].getTimestamp.equals(timestamp))
+ assert(dstreamAtTimestamp.checkpointAppName == customAppName)
+ assert(dstreamAtTimestamp.checkpointInterval == customCheckpointInterval)
+ assert(dstreamAtTimestamp._storageLevel == customStorageLevel)
+ assert(dstreamAtTimestamp.kinesisCreds == customKinesisCreds)
+ assert(dstreamAtTimestamp.dynamoDBCreds == Option(customDynamoDBCreds))
+ assert(dstreamAtTimestamp.cloudWatchCreds == Option(customCloudWatchCreds))
+ }
+
+ test("old Api should throw UnsupportedOperationExceptionexception with AT_TIMESTAMP") {
+ val streamName: String = "a-very-nice-stream-name"
+ val endpointUrl: String = "https://kinesis.us-west-2.amazonaws.com"
+ val region: String = "us-west-2"
+ val appName: String = "a-very-nice-kinesis-app"
+ val checkpointInterval: Duration = Seconds.apply(30)
+ val storageLevel: StorageLevel = StorageLevel.MEMORY_ONLY
+
+ // This should not build.
+ // InitialPositionInStream.AT_TIMESTAMP is not supported in old Api.
+ // The builder Api in KinesisInputDStream should be used.
+ intercept[UnsupportedOperationException] {
+ val kinesisDStream: KinesisInputDStream[Array[Byte]] = KinesisInputDStream.builder
+ .streamingContext(ssc)
+ .streamName(streamName)
+ .endpointUrl(endpointUrl)
+ .regionName(region)
+ .initialPositionInStream(InitialPositionInStream.AT_TIMESTAMP)
+ .checkpointAppName(appName)
+ .checkpointInterval(checkpointInterval)
+ .storageLevel(storageLevel)
+ .build
+ }
}
}
http://git-wip-us.apache.org/repos/asf/spark/blob/0e683300/external/kinesis-asl/src/test/scala/org/apache/spark/streaming/kinesis/KinesisStreamSuite.scala
----------------------------------------------------------------------
diff --git a/external/kinesis-asl/src/test/scala/org/apache/spark/streaming/kinesis/KinesisStreamSuite.scala b/external/kinesis-asl/src/test/scala/org/apache/spark/streaming/kinesis/KinesisStreamSuite.scala
index 7e5bda9..a7a68eb 100644
--- a/external/kinesis-asl/src/test/scala/org/apache/spark/streaming/kinesis/KinesisStreamSuite.scala
+++ b/external/kinesis-asl/src/test/scala/org/apache/spark/streaming/kinesis/KinesisStreamSuite.scala
@@ -34,6 +34,7 @@ import org.apache.spark.rdd.RDD
import org.apache.spark.storage.{StorageLevel, StreamBlockId}
import org.apache.spark.streaming._
import org.apache.spark.streaming.dstream.ReceiverInputDStream
+import org.apache.spark.streaming.kinesis.KinesisInitialPositions.Latest
import org.apache.spark.streaming.kinesis.KinesisReadConfigurations._
import org.apache.spark.streaming.kinesis.KinesisTestUtils._
import org.apache.spark.streaming.receiver.BlockManagerBasedStoreResult
@@ -178,7 +179,7 @@ abstract class KinesisStreamTests(aggregateTestData: Boolean) extends KinesisFun
.streamName(testUtils.streamName)
.endpointUrl(testUtils.endpointUrl)
.regionName(testUtils.regionName)
- .initialPositionInStream(InitialPositionInStream.LATEST)
+ .initialPosition(new Latest())
.checkpointInterval(Seconds(10))
.storageLevel(StorageLevel.MEMORY_ONLY)
.build()
@@ -209,7 +210,7 @@ abstract class KinesisStreamTests(aggregateTestData: Boolean) extends KinesisFun
.streamName(testUtils.streamName)
.endpointUrl(testUtils.endpointUrl)
.regionName(testUtils.regionName)
- .initialPositionInStream(InitialPositionInStream.LATEST)
+ .initialPosition(new Latest())
.checkpointInterval(Seconds(10))
.storageLevel(StorageLevel.MEMORY_ONLY)
.buildWithMessageHandler(addFive(_))
@@ -245,7 +246,7 @@ abstract class KinesisStreamTests(aggregateTestData: Boolean) extends KinesisFun
.streamName("dummyStream")
.endpointUrl(dummyEndpointUrl)
.regionName(dummyRegionName)
- .initialPositionInStream(InitialPositionInStream.LATEST)
+ .initialPosition(new Latest())
.checkpointInterval(Seconds(10))
.storageLevel(StorageLevel.MEMORY_ONLY)
.build()
@@ -293,7 +294,7 @@ abstract class KinesisStreamTests(aggregateTestData: Boolean) extends KinesisFun
.streamName(localTestUtils.streamName)
.endpointUrl(localTestUtils.endpointUrl)
.regionName(localTestUtils.regionName)
- .initialPositionInStream(InitialPositionInStream.LATEST)
+ .initialPosition(new Latest())
.checkpointInterval(Seconds(10))
.storageLevel(StorageLevel.MEMORY_ONLY)
.build()
@@ -369,7 +370,7 @@ abstract class KinesisStreamTests(aggregateTestData: Boolean) extends KinesisFun
.streamName(testUtils.streamName)
.endpointUrl(testUtils.endpointUrl)
.regionName(testUtils.regionName)
- .initialPositionInStream(InitialPositionInStream.LATEST)
+ .initialPosition(new Latest())
.checkpointInterval(Seconds(10))
.storageLevel(StorageLevel.MEMORY_ONLY)
.build()
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org