xiaolong-sn commented on a change in pull request #12920:
URL: https://github.com/apache/flink/pull/12920#discussion_r457122566
##########
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());
+ }
+ if (RecordPublisherType.valueOf(recordPublisherType) ==
RecordPublisherType.EFO) {
+ String efoRegistrationType;
+ if
(config.containsKey(ConsumerConfigConstants.EFO_REGISTRATION_TYPE)) {
+ efoRegistrationType =
config.getProperty(ConsumerConfigConstants.EFO_REGISTRATION_TYPE);
+ // specified efo registration type in
stream must be either LAZY, EAGER or NONE.
+ try {
+
EFORegistrationType.valueOf(efoRegistrationType);
+ } catch (IllegalArgumentException e) {
+ StringBuilder sb = new
StringBuilder();
+ for (EFORegistrationType ert :
EFORegistrationType.values()) {
+
sb.append(ert.toString()).append(", ");
+ }
+ throw new
IllegalArgumentException("Invalid efo registration type in stream set in
config. Valid values are: " + sb.toString());
+ }
+ } else {
+ efoRegistrationType =
EFORegistrationType.LAZY.toString();
+ }
+ if
(EFORegistrationType.valueOf(efoRegistrationType) == EFORegistrationType.NONE) {
+ //if the registration type is NONE,
then for each stream there must be an according consumer ARN
+ for (String stream : streams) {
+ String efoConsumerARNKey =
ConsumerConfigConstants.EFO_CONSUMER_ARN_PREFIX + "." + stream;
+ if
(!config.containsKey(efoConsumerARNKey)) {
+ throw new
IllegalArgumentException("Invalid efo consumer arn settings for not providing "
+ efoConsumerARNKey);
Review comment:
I've collected all the missing keys and given a single error message.
----------------------------------------------------------------
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]
