[FLINK-4582] [kinesis] Consuming data from DynamoDB streams to flink #6968

yxu-valleytider · 2018-10-30T23:45:05Z

What is the purpose of the change

This PR introduces a new Flink source to consume directly from dynamodb streams. This new source is built on top of the existing Kinesis connector. It interacts with the dynamodb streams via a dynamodb-streams-kinesis-adapter client.

Brief change log

New data stream can be constructed to directly pull data from DynamoDB streams.

DataStream<String> dynamodbStreams = env.addSource(new FlinkDynamodbStreamsConsumer(streamName, ..., ...))

Other changes include:

DynamodbStreamsProxy class which uses a dynamodbstreams-kinesis adapter client to interact with Dynamodb streams.
Port the describeStream API from the flink-1.5 branch into DynamodbStreamsProxy.
Minimalist DynamodbStreamsSchema class which helps deserialize dynamodb streams records

Verifying this change

This change is already covered by most of the existing Flink kinesis connector tests.

Manual tests are provided to verify that the dynamodbstreams connector can:

consume from a stream given its ARN
consume from a given table with new stream created

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (yes)
Added optional dependency on AWS dynamodb-streams-kinesis-adapter.
The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
The serializers: (no)
The runtime per-record code paths (performance sensitive): (no)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (no)
The S3 file system connector: (no)

Documentation

Does this pull request introduce a new feature? (yes)
If yes, how is the feature documented? (JavaDocs)

zentol · 2018-10-31T10:15:52Z

@yxu-valleytider Please make sure that the Pull Request title references the corresponding JIRA. I have modified the title accordingly this time.

tweise · 2018-10-31T13:14:35Z

@yxu-valleytider please include tag [kinesis] into the PR title and also format the commit messages as per convention [<JIRA>] [kinesis] add nice feature

tweise · 2018-10-31T13:18:23Z

...java/org/apache/flink/streaming/connectors/dynamodbstreams/FlinkDynamodbStreamsConsumer.java

+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.connectors.dynamodbstreams;


All of the newly added packages should be under org.apache.flink.streaming.connectors.kinesis as they are part of the flink-connector-kinesis module.

Hmm, both the dynamodbstreams and kinesis modules are under the parent module org.apache.flink.streaming.connectors. This is also consistent with the directory structure.

These are just packages with the module flink-connector-kinesis and the package prefix should reflect that.

In that case I don't mind putting the files into a separate directory, e.g., flink-connector-dynamodbstreams. WDYT ?

I don't think that will work: We had already determined in our internal work that we need a single shaded AWS dependency. I don't think it's desirable either, because it is just a slight variation of the Kinesis consumer that does not deserve a separate module.

tweise · 2018-10-31T13:19:25Z

.../apache/flink/streaming/connectors/dynamodbstreams/model/DynamodbStreamsShardHandleTest.java

+			DynamodbStreamsShardHandle.compareShardIds(shardIdInvalid, shardIdValid);
+			fail("invalid shard Id" + shardIdInvalid + " should trigger exception");
+		} catch (IllegalArgumentException e) {
+			// ignore


this should be // expected instead?

yes // expected clarifies more.

tweise · 2018-10-31T13:21:13Z

...-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxy.java

@@ -73,7 +73,7 @@
 	private static final Logger LOG = LoggerFactory.getLogger(KinesisProxy.class);

 	/** The actual Kinesis client from the AWS SDK that we will be using to make calls. */
-	private final AmazonKinesis kinesisClient;
+	protected final AmazonKinesis kinesisClient;


why this modifier change?

It is used inside DynamodbStreamsProxy to execute the describeStream call.

If describeStream moves to KinesisProxy, this change becomes unnecessary.

yxu-valleytider · 2018-11-01T07:45:22Z

PTAL @tzulitai

yxu-valleytider · 2018-11-22T19:51:59Z

Adjusted package name based on comments.

PTAL @tweise @tzulitai

tweise · 2018-11-25T18:55:24Z

...rc/main/java/org/apache/flink/streaming/connectors/kinesis/FlinkDynamodbStreamsConsumer.java

+		super(streams, deserializer, config);
+	}
+
+	public static <T> FlinkDynamodbStreamsConsumer<T> create(String stream,


Please remove these static create methods, they don't provide any benefit since the constructors are public (which is consistent with the base class).

tweise · 2018-11-25T19:00:46Z

.../src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/DynamodbStreamsProxy.java

+	 *
+	 * @return the result of the describe stream operation
+	 */
+	private DescribeStreamResult describeStream(String streamName, @Nullable String startShardId)


Unless there is anything specific to DynamoDB in this implementation, move it to KinesisProxy. (DescribeStream is standard Kinesis, and it was used for discovery before listShards got introduced.)

OK to move. Just in that case DescribeStream will be marked as protected at least, in order to be visible to DynamoDBStreamsProxy.

By keeping it inside the DynamodbStreamsProxy layer this logic can be private specific for interacting with DynamodbStreams.

@tweise

Making it protected is fine. Since these are common capabilities of the Kinesis API it is better to keep it in one place.

tweise · 2018-11-25T19:01:57Z

...-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxy.java

@@ -73,7 +73,7 @@
 	private static final Logger LOG = LoggerFactory.getLogger(KinesisProxy.class);

 	/** The actual Kinesis client from the AWS SDK that we will be using to make calls. */
-	private final AmazonKinesis kinesisClient;
+	protected final AmazonKinesis kinesisClient;


If describeStream moves to KinesisProxy, this change becomes unnecessary.

tweise · 2018-11-25T19:06:36Z

.../main/java/org/apache/flink/streaming/connectors/kinesis/config/ConsumerConfigConstants.java

+	 * Different tag name to distinguish from "flink.stream.describe.backoff.base"
+	 * since the latter is deprecated.
+	 */
+	public static final String DYNAMODB_STREAMS_DESCRIBE_BACKOFF_BASE =


Remove these and remove the @Deprecated annotations from the prior properties.

@tweise are you ok with removing this function KinesisProxy::replaceDeprecatedConsumerKeys as well ?

It would be good to modify the method to map/replicate the keys but not remove them (since you need them for dynamo). Add a comment that this is for backward compatibility for the regular proxy that is now using listShards instead of describeShards. We can remove the mapping in later release.

tweise · 2018-11-25T19:12:19Z

...java/org/apache/flink/streaming/connectors/kinesis/internals/DynamodbStreamsDataFetcher.java

+	 * Updates the last discovered shard of a subscribed stream; only updates if the update is valid.
+	 */
+	@Override
+	public void advanceLastDiscoveredShardOfStream(String stream, String shardId) {


Rather than duplicating complete logic from the base class, can we just extract what is unique to DynamoDB? That might also eliminate the need to expose subscribedStreamsToLastDiscoveredShardIds?

yxu-valleytider · 2018-12-21T01:43:57Z

PTAL @tweise

tweise · 2018-12-21T01:52:01Z

.../src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/DynamodbStreamsProxy.java

+		super(configProps);
+	}
+
+


Remove extra line.

tweise · 2018-12-21T01:56:16Z

...-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxy.java

@@ -55,6 +58,12 @@
 import java.util.Properties;
 import java.util.Random;

+import static org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants.DEFAULT_STREAM_DESCRIBE_BACKOFF_BASE;


Let's stay consistent with the existing implementation and not add these imports.

tweise · 2018-12-21T02:02:41Z

.../main/java/org/apache/flink/streaming/connectors/kinesis/config/ConsumerConfigConstants.java

+	 * Boolean to indicate whether to compare/enforce shardId format based on the one defined in
+	 * DynamodbStreamsShardHandle.
+	 */
+	public static final String DYNAMODB_STREAMS_SHARDID_FORMAT_CHECK =


This property isn't actually used? If so, please remove.

The property is currently used inside DynamoDBStreamsDataFetcher.

The key is never used to set the property, that's what not used refers to. The documentation also does not mention why anyone would use it. If there is no use case for this, let's remove it.

changed the default to true and removed the config property.

tweise · 2018-12-21T02:04:22Z

Does the PR need to be rebased?

tweise · 2018-12-21T02:10:42Z

.../src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/DynamodbStreamsProxy.java

+/**
+ * DynamoDB streams proxy: interface interacting with the DynamoDB streams.
+ */
+public class DynamodbStreamsProxy extends KinesisProxy {


DynamodbStreamsProxy => DynamoDBStreamsProxy (same may apply elsewhere also)

tweise · 2018-12-21T02:12:22Z

...-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxy.java

@@ -469,6 +496,61 @@ private ListShardsResult listShards(String streamName, @Nullable String startSha
 		return listShardsResults;
 	}

+	/**
+	 * Get metainfo for a Dynamodb stream, which contains information about which shards this


Remove DynamoDB reference here. This is applicable to proper Kinesis also.

tweise · 2018-12-21T02:15:06Z

...esis/src/main/java/org/apache/flink/streaming/connectors/kinesis/util/KinesisConfigUtil.java

-				configProps.setProperty(newKey, configProps.getProperty(deprecatedOldKey));
-				configProps.remove(deprecatedOldKey);
+			if (configProps.containsKey(oldKey)) {
+				LOG.warn("Backfill the property key {} based on the original key {}", newKey, oldKey);


Remove this warning since it isn't applicable any more (the key is expected to be used for DynamoDB).

tweise · 2018-12-21T02:16:38Z

...esis/src/main/java/org/apache/flink/streaming/connectors/kinesis/util/KinesisConfigUtil.java

+	 *
+	 * In the context of 1), the set of configurations needs to be translated to the corresponding
+	 * configurations in the Kinesis listShards API. In the mean time, keep these configs since
+	 * they may be useful in the context of 2), i.e., polling data from a DynamoDB stream.


"may be useful" => "are applicable"

tweise · 2018-12-21T18:18:21Z

Dependency diff shows the following:

< [INFO] |  |  +- com.fasterxml.jackson.core:jackson-databind:jar:2.6.7.1:compile
< [INFO] |  |  |  +- com.fasterxml.jackson.core:jackson-annotations:jar:2.6.0:compile
< [INFO] |  |  |  \- com.fasterxml.jackson.core:jackson-core:jar:2.6.7:compile
79a77,79
> [INFO] |  +- com.amazonaws:aws-java-sdk-dynamodb:jar:1.11.272:compile
> [INFO] |  |  \- com.amazonaws:aws-java-sdk-s3:jar:1.11.272:compile
> [INFO] |  |     \- com.amazonaws:aws-java-sdk-kms:jar:1.11.272:compile
80a81,84
> [INFO] +- com.amazonaws:dynamodb-streams-kinesis-adapter:jar:1.4.0:compile
> [INFO] |  \- com.fasterxml.jackson.core:jackson-databind:jar:2.6.6:compile
> [INFO] |     +- com.fasterxml.jackson.core:jackson-annotations:jar:2.6.0:compile
> [INFO] |     \- com.fasterxml.jackson.core:jackson-core:jar:2.6.6:compile

The version of com.fasterxml.jackson.core:jackson-databind that comes with com.amazonaws:dynamodb-streams-kinesis-adapter is 2.6.6 vs. 2.6.7 from com.amazonaws:aws-java-sdk-core previously.

…sumer

…ents

tweise · 2018-12-22T16:20:56Z

@yxu-valleytider thanks for the contribution!

…pache#6968) Introduces a new Flink source to consume from DynamoDB streams. This new source is built on top of the existing Kinesis connector. It interacts with the DynamoDB streams via a dynamodb-streams-kinesis-adapter client.

zentol changed the title ~~Consuming data from DynamoDB streams to flink~~ [FLINK-4582] Consuming data from DynamoDB streams to flink Oct 31, 2018

tweise reviewed Oct 31, 2018

View reviewed changes

yxu-valleytider changed the title ~~[FLINK-4582] Consuming data from DynamoDB streams to flink~~ [FLINK-4582] [kinesis] Consuming data from DynamoDB streams to flink Oct 31, 2018

tweise requested changes Nov 25, 2018

View reviewed changes

tweise requested changes Dec 21, 2018

View reviewed changes

tweise reviewed Dec 21, 2018

View reviewed changes

tweise approved these changes Dec 22, 2018

View reviewed changes

yxu-valleytider added 11 commits December 21, 2018 21:38

FLINK-4582: initial commit for flink dynamodbstreams connector

22a2b40

FLINK-4582: small format fix

03029cc

Fix ConsumeFromDynamodbStreams example to use FlinkDynamodbStreamsCon…

a323e55

…sumer

address one review comment

6c26109

FLINK-4582: adjust package name based on review comment

06d7a5c

fix checkstyle

43f5ec5

minor naming adjustment

e6f300f

FLINK-4582: fix compile and checkstyle

2f94fe7

FLINK-4582: move describeStream into KinesisProxy; other review commm…

073470b

…ents

FLINK-4582: more review comments

c08e9b2

FLINK-4582: remove config option; exclude older databind dependency

45f42e4

tweise added 3 commits December 21, 2018 21:41

Update ConsumerConfigConstants.java

ab5368e

Update FlinkDynamoDBStreamsConsumer.java

2195199

Update KinesisProxy.java

d43ba9e

tweise force-pushed the dynamodbstreams-connector branch from acbfa2e to d43ba9e Compare December 22, 2018 05:43

fixup

e89b4a5

tweise merged commit bc3eb82 into apache:master Dec 22, 2018

rmetzger added component=Connectors/Common component=Connectors/Kinesis labels Mar 18, 2019

[FLINK-4582] [kinesis] Consuming data from DynamoDB streams to flink #6968

[FLINK-4582] [kinesis] Consuming data from DynamoDB streams to flink #6968

Conversation

yxu-valleytider commented Oct 30, 2018

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

zentol commented Oct 31, 2018

tweise commented Oct 31, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

yxu-valleytider commented Nov 1, 2018

yxu-valleytider commented Nov 22, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

yxu-valleytider Dec 20, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

yxu-valleytider commented Dec 21, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tweise commented Dec 21, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tweise commented Dec 21, 2018

tweise commented Dec 22, 2018

yxu-valleytider Dec 20, 2018 •

edited

Loading