[Hotfix][OptionRule] Fix option rule about all connectors

seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/util/ConfigValidator.java

@@ -108,30 +108,24 @@ boolean validate(RequiredOption.BundledRequiredOptions bundledRequiredOptions) {
     }
 
     void validate(RequiredOption.ExclusiveRequiredOptions exclusiveRequiredOptions) {
-        List<RequiredOption.BundledRequiredOptions> presentBundledRequiredOptions = new ArrayList<>();
         List<Option<?>> presentOptions = new ArrayList<>();
-        for (RequiredOption.BundledRequiredOptions bundledOptions : exclusiveRequiredOptions.getExclusiveBundledOptions()) {
-            if (validate(bundledOptions)) {
-                presentBundledRequiredOptions.add(bundledOptions);
-            }
-        }
 
         for (Option<?> option : exclusiveRequiredOptions.getExclusiveOptions()) {
             if (hasOption(option)) {
                 presentOptions.add(option);
             }
         }
-        int count = presentBundledRequiredOptions.size() + presentOptions.size();
+        int count = presentOptions.size();
         if (count == 1) {
             return;
         }
         if (count == 0) {
             throw new OptionValidationException("There are unconfigured options, these options(%s) are mutually exclusive, allowing only one set(\"[] for a set\") of options to be configured.",
-                getOptionKeys(exclusiveRequiredOptions.getExclusiveOptions(), exclusiveRequiredOptions.getExclusiveBundledOptions()));
+                getOptionKeys(exclusiveRequiredOptions.getExclusiveOptions()));
         }
         if (count > 1) {
             throw new OptionValidationException("These options(%s) are mutually exclusive, allowing only one set(\"[] for a set\") of options to be configured.",
-                getOptionKeys(presentOptions, presentBundledRequiredOptions));
+                getOptionKeys(presentOptions));
         }
     }
 

seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/util/OptionRule.java

@@ -19,10 +19,13 @@
 
 import org.apache.seatunnel.api.configuration.Option;
 
+import lombok.NonNull;
+
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Objects;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
  * Validation rule for {@link Option}.
@@ -127,16 +130,20 @@ private Builder() {
          * <p> This options will not be validated.
          * <p> This is used by the web-UI to show what options are available.
          */
-        public Builder optional(Option<?>... options) {
+        public Builder optional(@NonNull Option<?>... options) {
+            for (Option<?> option : options) {
+                verifyDuplicate(option, "OptionsOption");
+            }
             this.optionalOptions.addAll(Arrays.asList(options));
             return this;
         }
 
         /**
          * Absolutely required options without any constraints.
          */
-        public Builder required(Option<?>... options) {
+        public Builder required(@NonNull Option<?>... options) {
             for (Option<?> option : options) {
+                verifyDuplicate(option, "RequiredOption");
                 verifyRequiredOptionDefaultValue(option);
             }
             this.requiredOptions.add(RequiredOption.AbsolutelyRequiredOptions.of(options));
@@ -146,43 +153,59 @@ public Builder required(Option<?>... options) {
         /**
          * Exclusive options, only one of the options needs to be configured.
          */
-        public Builder exclusive(Option<?>... options) {
+        public Builder exclusive(@NonNull Option<?>... options) {
             if (options.length <= 1) {
                 throw new OptionValidationException("The number of exclusive options must be greater than 1.");
             }
             for (Option<?> option : options) {
+                verifyDuplicate(option, "ExclusiveOption");
                 verifyRequiredOptionDefaultValue(option);
             }
             this.requiredOptions.add(RequiredOption.ExclusiveRequiredOptions.of(options));
             return this;
         }
 
-        public Builder exclusive(RequiredOption.ExclusiveRequiredOptions exclusiveRequiredOptions) {
-            this.requiredOptions.add(exclusiveRequiredOptions);
-            return this;
-        }
+        public <T> Builder conditional(@NonNull Option<T> conditionalOption, @NonNull List<T> expectValues, @NonNull Option<?>... requiredOptions) {
+            for (Option<?> o : requiredOptions) {
+                verifyDuplicate(o, "ConditionalOption");
+                verifyRequiredOptionDefaultValue(o);
+            }
 
-        /**
-         * Conditional options, These options are required if the {@link Option} == expectValue.
-         */
-        public <T> Builder conditional(Option<T> option, T expectValue, Option<?>... requiredOptions) {
-            return conditional(Condition.of(option, expectValue), requiredOptions);
-        }
+            verifyConditionalExists(conditionalOption);
 
-        /**
-         * Conditional options, These options are required if the {@link Condition} evaluates to true.
-         */
-        public Builder conditional(Condition<?> condition, Option<?>... requiredOptions) {
-            return conditional(Expression.of(condition), requiredOptions);
+            if (expectValues.size() == 0) {
+                throw new OptionValidationException(
+                    String.format("conditional option '%s' must have expect values .", conditionalOption.key()));
+            }
+
+            /**
+             * Each parameter can only be controlled by one other parameter
+             */
+            Expression expression = Expression.of(Condition.of(conditionalOption, expectValues.get(0)));
+            for (int i = 0; i < expectValues.size(); i++) {
+                if (i != 0) {
+                    expression = expression.or(Expression.of(Condition.of(conditionalOption, expectValues.get(i))));
+                }
+            }
+
+            this.requiredOptions.add(RequiredOption.ConditionalRequiredOptions.of(expression,
+                new ArrayList<>(Arrays.asList(requiredOptions))));
+            return this;
         }
 
-        /**
-         * Conditional options, These options are required if the {@link Expression} evaluates to true.
-         */
-        public Builder conditional(Expression expression, Option<?>... requiredOptions) {
+        public <T> Builder conditional(@NonNull Option<T> conditionalOption, @NonNull T expectValue, @NonNull Option<?>... requiredOptions) {
             for (Option<?> o : requiredOptions) {
+                verifyDuplicate(o, "ConditionalOption");
                 verifyRequiredOptionDefaultValue(o);
             }
+
+            verifyConditionalExists(conditionalOption);
+
+            /**
+             * Each parameter can only be controlled by one other parameter
+             */
+            Expression expression = Expression.of(Condition.of(conditionalOption, expectValue));
+
             this.requiredOptions.add(RequiredOption.ConditionalRequiredOptions.of(expression,
                 new ArrayList<>(Arrays.asList(requiredOptions))));
             return this;
@@ -191,7 +214,10 @@ public Builder conditional(Expression expression, Option<?>... requiredOptions)
         /**
          * Bundled options, must be present or absent together.
          */
-        public Builder bundled(Option<?>... requiredOptions) {
+        public Builder bundled(@NonNull Option<?>... requiredOptions) {
+            for (Option<?> option : requiredOptions) {
+                verifyDuplicate(option, "BundledOption");
+            }
             this.requiredOptions.add(RequiredOption.BundledRequiredOptions.of(requiredOptions));
             return this;
         }
@@ -200,11 +226,40 @@ public OptionRule build() {
             return new OptionRule(optionalOptions, requiredOptions);
         }
 
-        private void verifyRequiredOptionDefaultValue(Option<?> option) {
+        private void verifyRequiredOptionDefaultValue(@NonNull Option<?> option) {
             if (option.defaultValue() != null) {
                 throw new OptionValidationException(
                     String.format("Required option '%s' should have no default value.", option.key()));
             }
         }
+
+        private void verifyDuplicate(@NonNull Option<?> option, @NonNull String currentOptionType) {
+            if (optionalOptions.contains(option)) {
+                throw new OptionValidationException(
+                    String.format("%s '%s' duplicate in option options.", currentOptionType, option.key()));
+            }
+
+            requiredOptions.forEach(requiredOption -> {
+                if (requiredOption.getOptions().contains(option)) {
+                    throw new OptionValidationException(
+                        String.format("%s '%s' duplicate in '%s'.", currentOptionType, option.key(), requiredOption.getClass().getName()));
+                }
+            });
+        }
+
+        private void verifyConditionalExists(@NonNull Option<?> option) {
+            boolean inOptions = optionalOptions.contains(option);
+            AtomicBoolean inRequired = new AtomicBoolean(false);
+            requiredOptions.forEach(requiredOption -> {
+                if (requiredOption.getOptions().contains(option)) {
+                    inRequired.set(true);
+                }
+            });
+
+            if (!inOptions && !inRequired.get()) {
+                throw new OptionValidationException(
+                    String.format("Conditional '%s' not found in options.", option.key()));
+            }
+        }
     }
 }

seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/util/RequiredOption.java

@@ -22,6 +22,7 @@
 import org.apache.seatunnel.api.configuration.Option;
 
 import lombok.Getter;
+import lombok.NonNull;
 
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -30,25 +31,21 @@
 
 public interface RequiredOption {
 
+    List<Option<?>> getOptions();
+
     /**
      * These options are mutually exclusive, allowing only one set of options to be configured.
      */
     @Getter
     class ExclusiveRequiredOptions implements RequiredOption {
-        private final List<BundledRequiredOptions> exclusiveBundledOptions;
         private final List<Option<?>> exclusiveOptions;
 
-        ExclusiveRequiredOptions(List<BundledRequiredOptions> exclusiveBundledOptions, List<Option<?>> exclusiveOptions) {
-            this.exclusiveBundledOptions = exclusiveBundledOptions;
+        public ExclusiveRequiredOptions(@NonNull List<Option<?>> exclusiveOptions) {
             this.exclusiveOptions = exclusiveOptions;
         }
 
-        public static ExclusiveRequiredOptions of(Option<?>... exclusiveOptions) {
-            return ExclusiveRequiredOptions.of(new ArrayList<>(), exclusiveOptions);
-        }
-
-        public static ExclusiveRequiredOptions of(List<BundledRequiredOptions> exclusiveBundledOptions, Option<?>... exclusiveOptions) {
-            return new ExclusiveRequiredOptions(exclusiveBundledOptions, new ArrayList<>(Arrays.asList(exclusiveOptions)));
+        public static ExclusiveRequiredOptions of(Option<?>... options) {
+            return new ExclusiveRequiredOptions(new ArrayList<>(Arrays.asList(options)));
         }
 
         @Override
@@ -65,12 +62,17 @@ public boolean equals(Object obj) {
 
         @Override
         public int hashCode() {
-            return Objects.hash(exclusiveBundledOptions, exclusiveOptions);
+            return Objects.hash(exclusiveOptions);
         }
 
         @Override
         public String toString() {
-            return String.format("Exclusive required set options: %s", getOptionKeys(exclusiveOptions, exclusiveBundledOptions));
+            return String.format("Exclusive required set options: %s", getOptionKeys(exclusiveOptions));
+        }
+
+        @Override
+        public List<Option<?>> getOptions() {
+            return exclusiveOptions;
         }
     }
 
@@ -110,6 +112,11 @@ public int hashCode() {
         public String toString() {
             return String.format("Absolutely required options: '%s'", getOptionKeys(requiredOption));
         }
+
+        @Override
+        public List<Option<?>> getOptions() {
+            return requiredOption;
+        }
     }
 
     class ConditionalRequiredOptions implements RequiredOption {
@@ -158,6 +165,11 @@ public int hashCode() {
         public String toString() {
             return String.format("Condition expression: %s, Required options: %s", expression, getOptionKeys(requiredOption));
         }
+
+        @Override
+        public List<Option<?>> getOptions() {
+            return requiredOption;
+        }
     }
 
     /**
@@ -203,5 +215,10 @@ public int hashCode() {
         public String toString() {
             return String.format("Bundled Required options: %s", getOptionKeys(requiredOption));
         }
+
+        @Override
+        public List<Option<?>> getOptions() {
+            return requiredOption;
+        }
     }
 }