You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by fh...@apache.org on 2019/04/16 10:06:46 UTC
[flink] 05/06: [FLINK-12137][docs] Use AWSConfigConstants instead
of ConsumerConfigConstants in examples.
This is an automated email from the ASF dual-hosted git repository.
fhueske pushed a commit to branch release-1.8
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 7c8e98069be8bba2c12f471f2fc24ef792e4addc
Author: joongkeun.yang <yj...@gmail.com>
AuthorDate: Tue Apr 9 17:34:31 2019 +0900
[FLINK-12137][docs] Use AWSConfigConstants instead of ConsumerConfigConstants in examples.
This closes #8128.
---
docs/dev/connectors/kinesis.md | 17 +++++++++--------
1 file changed, 9 insertions(+), 8 deletions(-)
diff --git a/docs/dev/connectors/kinesis.md b/docs/dev/connectors/kinesis.md
index 5e0b2f3..f39da3d 100644
--- a/docs/dev/connectors/kinesis.md
+++ b/docs/dev/connectors/kinesis.md
@@ -82,9 +82,9 @@ Before consuming data from Kinesis streams, make sure that all streams are creat
<div data-lang="java" markdown="1">
{% highlight java %}
Properties consumerConfig = new Properties();
-consumerConfig.put(ConsumerConfigConstants.AWS_REGION, "us-east-1");
-consumerConfig.put(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "aws_access_key_id");
-consumerConfig.put(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "aws_secret_access_key");
+consumerConfig.put(AWSConfigConstants.AWS_REGION, "us-east-1");
+consumerConfig.put(AWSConfigConstants.AWS_ACCESS_KEY_ID, "aws_access_key_id");
+consumerConfig.put(AWSConfigConstants.AWS_SECRET_ACCESS_KEY, "aws_secret_access_key");
consumerConfig.put(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "LATEST");
StreamExecutionEnvironment env = StreamExecutionEnvironment.getEnvironment();
@@ -96,9 +96,9 @@ DataStream<String> kinesis = env.addSource(new FlinkKinesisConsumer<>(
<div data-lang="scala" markdown="1">
{% highlight scala %}
val consumerConfig = new Properties()
-consumerConfig.put(ConsumerConfigConstants.AWS_REGION, "us-east-1")
-consumerConfig.put(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "aws_access_key_id")
-consumerConfig.put(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "aws_secret_access_key")
+consumerConfig.put(AWSConfigConstants.AWS_REGION, "us-east-1")
+consumerConfig.put(AWSConfigConstants.AWS_ACCESS_KEY_ID, "aws_access_key_id")
+consumerConfig.put(AWSConfigConstants.AWS_SECRET_ACCESS_KEY, "aws_secret_access_key")
consumerConfig.put(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "LATEST")
val env = StreamExecutionEnvironment.getEnvironment
@@ -110,10 +110,11 @@ val kinesis = env.addSource(new FlinkKinesisConsumer[String](
</div>
The above is a simple example of using the consumer. Configuration for the consumer is supplied with a `java.util.Properties`
-instance, the configuration keys for which can be found in `ConsumerConfigConstants`. The example
+instance, the configuration keys for which can be found in `AWSConfigConstants` (AWS-specific parameters) and
+`ConsumerConfigConstants` (Kinesis consumer parameters). The example
demonstrates consuming a single Kinesis stream in the AWS region "us-east-1". The AWS credentials are supplied using the basic method in which
the AWS access key ID and secret access key are directly supplied in the configuration (other options are setting
-`ConsumerConfigConstants.AWS_CREDENTIALS_PROVIDER` to `ENV_VAR`, `SYS_PROP`, `PROFILE`, `ASSUME_ROLE`, and `AUTO`). Also, data is being consumed
+`AWSConfigConstants.AWS_CREDENTIALS_PROVIDER` to `ENV_VAR`, `SYS_PROP`, `PROFILE`, `ASSUME_ROLE`, and `AUTO`). Also, data is being consumed
from the newest position in the Kinesis stream (the other option will be setting `ConsumerConfigConstants.STREAM_INITIAL_POSITION`
to `TRIM_HORIZON`, which lets the consumer start reading the Kinesis stream from the earliest record possible).