xiaolong-sn commented on a change in pull request #12920:
URL: https://github.com/apache/flink/pull/12920#discussion_r457122192
##########
File path:
flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/util/KinesisConfigUtil.java
##########
@@ -169,6 +232,55 @@ public static void
validateConsumerConfiguration(Properties config) {
}
}
+ public static void validateEFOConfiguration(Properties config,
List<String> streams) {
+ if
(config.containsKey(ConsumerConfigConstants.RECORD_PUBLISHER_TYPE)) {
+ String recordPublisherType =
config.getProperty(ConsumerConfigConstants.RECORD_PUBLISHER_TYPE);
+
+ // specified record publisher type in stream must be
either EFO or POLLING
+ try {
+
RecordPublisherType.valueOf(recordPublisherType);
+ } catch (IllegalArgumentException e) {
+ StringBuilder sb = new StringBuilder();
+ for (RecordPublisherType rp :
RecordPublisherType.values()) {
+ sb.append(rp.toString()).append(", ");
+ }
+ throw new IllegalArgumentException("Invalid
record publisher type in stream set in config. Valid values are: " +
sb.toString());
Review comment:
I agree. I've moved this part to a new method.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
