[
https://issues.apache.org/jira/browse/FLINK-3229?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15253839#comment-15253839
]
ASF GitHub Bot commented on FLINK-3229:
---------------------------------------
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/1911#discussion_r60731207
--- Diff:
flink-streaming-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/config/KinesisConfigConstants.java
---
@@ -0,0 +1,51 @@
+package org.apache.flink.streaming.connectors.kinesis.config;
+
+/**
+ *
+ */
+public class KinesisConfigConstants {
+
+ //
------------------------------------------------------------------------
+ // Configuration Keys
+ //
------------------------------------------------------------------------
+
+ /** The max retries to retrieve metadata from a Kinesis stream using
describeStream API
+ * (Note: describeStream attempts may be temporarily blocked due to AWS
capping 5 attempts per sec) */
+ public static final String CONFIG_STREAM_DESCRIBE_RETRIES =
"flink.stream.describe.retry";
+
+ /** The backoff time between each describeStream attempt */
+ public static final String CONFIG_STREAM_DESCRIBE_BACKOFF =
"flink.stream.describe.backoff";
+
+ /** The initial position to start reading Kinesis streams from (LATEST
is used if not set) */
+ public static final String CONFIG_STREAM_INIT_POSITION_TYPE =
"flink.stream.initpos.type";
+
+ /** The credential provider type to use when AWS credentials are
required (BASIC is used if not set)*/
+ public static final String CONFIG_AWS_CREDENTIALS_PROVIDER_TYPE =
"aws.credentials.provider";
+
+ /** The AWS access key ID to use when setting credentials provider type
to BASIC */
+ public static final String
CONFIG_AWS_CREDENTIALS_PROVIDER_BASIC_ACCESSKEYID =
"aws.credentials.provider.basic.accesskeyid";
+
+ /** The AWS secret key to use when setting credentials provider type to
BASIC */
+ public static final String
CONFIG_AWS_CREDENTIALS_PROVIDER_BASIC_SECRETKEY =
"aws.credentials.provider.basic.secretkey";
+
+ /** Optional configuration for profile path if credential provider type
is set to be PROFILE */
+ public static final String CONFIG_AWS_CREDENTIALS_PROVIDER_PROFILE_PATH
= "aws.credentials.provider.profile.path";
+
+ /** Optional configuration for profile name if credential provider type
is set to be PROFILE */
+ public static final String CONFIG_AWS_CREDENTIALS_PROVIDER_PROFILE_NAME
= "aws.credentials.provider.profile.name";
+
+ /** The AWS region of the Kinesis streams to be pulled ("us-east-1" is
used if not set) */
+ public static final String CONFIG_AWS_REGION = "aws.region";
+
+
+ //
------------------------------------------------------------------------
+ // Default configuration values
+ //
------------------------------------------------------------------------
+
+ public static final String DEFAULT_AWS_REGION = "us-east-1";
--- End diff --
If we make the region a required argument, we won't need this anymore.
> Kinesis streaming consumer with integration of Flink's checkpointing mechanics
> ------------------------------------------------------------------------------
>
> Key: FLINK-3229
> URL: https://issues.apache.org/jira/browse/FLINK-3229
> Project: Flink
> Issue Type: Sub-task
> Components: Streaming Connectors
> Affects Versions: 1.0.0
> Reporter: Tzu-Li (Gordon) Tai
> Assignee: Tzu-Li (Gordon) Tai
>
> Opening a sub-task to implement data source consumer for Kinesis streaming
> connector (https://issues.apache.org/jira/browser/FLINK-3211).
> An example of the planned user API for Flink Kinesis Consumer:
> {code}
> Properties kinesisConfig = new Properties();
> config.put(KinesisConfigConstants.CONFIG_AWS_REGION, "us-east-1");
> config.put(KinesisConfigConstants.CONFIG_AWS_CREDENTIALS_PROVIDER_TYPE,
> "BASIC");
> config.put(
> KinesisConfigConstants.CONFIG_AWS_CREDENTIALS_PROVIDER_BASIC_ACCESSKEYID,
> "aws_access_key_id_here");
> config.put(
> KinesisConfigConstants.CONFIG_AWS_CREDENTIALS_PROVIDER_BASIC_SECRETKEY,
> "aws_secret_key_here");
> config.put(KinesisConfigConstants.CONFIG_STREAM_INIT_POSITION_TYPE,
> "LATEST"); // or TRIM_HORIZON
> DataStream<T> kinesisRecords = env.addSource(new FlinkKinesisConsumer<>(
> "kinesis_stream_name",
> new SimpleStringSchema(),
> kinesisConfig));
> {code}
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
