karubian commented on code in PR #165:
URL:
https://github.com/apache/flink-connector-aws/pull/165#discussion_r1765006662
##########
flink-connector-aws/flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/source/config/KinesisSourceConfigOptions.java:
##########
@@ -33,28 +36,35 @@ public enum InitialPosition {
}
public static final ConfigOption<InitialPosition> STREAM_INITIAL_POSITION =
- ConfigOptions.key("flink.stream.initpos")
+ ConfigOptions.key("kinesis.stream.init.position")
.enumType(InitialPosition.class)
.defaultValue(InitialPosition.LATEST)
.withDescription("The initial position to start reading
Kinesis streams.");
public static final ConfigOption<String> STREAM_INITIAL_TIMESTAMP =
- ConfigOptions.key("flink.stream.initpos.timestamp")
+ ConfigOptions.key("kinesis.stream.init.timestamp")
.stringType()
.noDefaultValue()
.withDescription(
"The initial timestamp at which to start reading
from the Kinesis stream. This is used when AT_TIMESTAMP is configured for the
STREAM_INITIAL_POSITION.");
public static final ConfigOption<String> STREAM_TIMESTAMP_DATE_FORMAT =
- ConfigOptions.key("flink.stream.initpos.timestamp.format")
+ ConfigOptions.key("kinesis.stream.init.timestamp.format")
.stringType()
.defaultValue("yyyy-MM-dd'T'HH:mm:ss.SSSXXX")
.withDescription(
"The date format used to parse the initial
timestamp at which to start reading from the Kinesis stream. This is used when
AT_TIMESTAMP is configured for the STREAM_INITIAL_POSITION.");
- public static final ConfigOption<Long> SHARD_DISCOVERY_INTERVAL_MILLIS =
- ConfigOptions.key("flink.shard.discovery.intervalmillis")
- .longType()
- .defaultValue(10000L)
+ public static final ConfigOption<Duration> SHARD_DISCOVERY_INTERVAL =
+ ConfigOptions.key("kinesis.shard.discovery.interval")
+ .durationType()
+ .defaultValue(Duration.ofSeconds(10))
.withDescription("The interval between each attempt to
discover new shards.");
+
+ public static final ConfigOption<Integer> SHARD_GET_RECORDS_MAX =
+ ConfigOptions.key("kinesis.shard.get-records.max-record-count")
+ .intType()
+ .defaultValue(10000)
+ .withDescription(
+ "The maximum number of getRecords attempts if we
get a recoverable exception.");
Review Comment:
The old connector has this as configuration.
[https://github.com/apache/flink-connector-aws/blob/main/flink-connector-aws/flink-[…]treaming/connectors/kinesis/config/ConsumerConfigConstants.java](https://github.com/apache/flink-connector-aws/blob/main/flink-connector-aws/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/config/ConsumerConfigConstants.java#L245)
I will update the description.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
