diff --git a/seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/util/ConfigValidator.java b/seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/util/ConfigValidator.java index 4041dc56da6..d6074173b94 100644 --- a/seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/util/ConfigValidator.java +++ b/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 presentBundledRequiredOptions = new ArrayList<>(); List> 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)); } } diff --git a/seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/util/OptionRule.java b/seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/util/OptionRule.java index 5b07e00e34f..42b9c44e343 100644 --- a/seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/util/OptionRule.java +++ b/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,7 +130,10 @@ private Builder() { *

This options will not be validated. *

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; } @@ -135,8 +141,9 @@ public Builder optional(Option... options) { /** * 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 Builder conditional(@NonNull Option conditionalOption, @NonNull List 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 Builder conditional(Option 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 Builder conditional(@NonNull Option 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())); + } + } } } diff --git a/seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/util/RequiredOption.java b/seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/util/RequiredOption.java index 85186aea2ba..d04e24a32aa 100644 --- a/seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/util/RequiredOption.java +++ b/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> getOptions(); + /** * These options are mutually exclusive, allowing only one set of options to be configured. */ @Getter class ExclusiveRequiredOptions implements RequiredOption { - private final List exclusiveBundledOptions; private final List> exclusiveOptions; - ExclusiveRequiredOptions(List exclusiveBundledOptions, List> exclusiveOptions) { - this.exclusiveBundledOptions = exclusiveBundledOptions; + public ExclusiveRequiredOptions(@NonNull List> exclusiveOptions) { this.exclusiveOptions = exclusiveOptions; } - public static ExclusiveRequiredOptions of(Option... exclusiveOptions) { - return ExclusiveRequiredOptions.of(new ArrayList<>(), exclusiveOptions); - } - - public static ExclusiveRequiredOptions of(List 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> 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> 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> 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> getOptions() { + return requiredOption; + } } } diff --git a/seatunnel-api/src/test/java/org/apache/seatunnel/api/configuration/util/ConfigValidatorTest.java b/seatunnel-api/src/test/java/org/apache/seatunnel/api/configuration/util/ConfigValidatorTest.java index 8e172ff0495..ba9acec79b3 100644 --- a/seatunnel-api/src/test/java/org/apache/seatunnel/api/configuration/util/ConfigValidatorTest.java +++ b/seatunnel-api/src/test/java/org/apache/seatunnel/api/configuration/util/ConfigValidatorTest.java @@ -34,9 +34,8 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.function.Executable; -import java.util.ArrayList; +import java.util.Arrays; import java.util.HashMap; -import java.util.List; import java.util.Map; public class ConfigValidatorTest { @@ -120,7 +119,7 @@ public void testSimpleExclusiveRequiredOptions() { Executable executable = () -> validate(config, rule); // all absent - assertEquals("ErrorCode:[API-02], ErrorDescription:[Option item validate failed] - There are unconfigured options, these options(['option.topic-pattern'], ['option.topic']) are mutually exclusive," + + assertEquals("ErrorCode:[API-02], ErrorDescription:[Option item validate failed] - There are unconfigured options, these options('option.topic-pattern', 'option.topic') are mutually exclusive," + " allowing only one set(\"[] for a set\") of options to be configured.", assertThrows(OptionValidationException.class, executable).getMessage()); @@ -130,46 +129,32 @@ public void testSimpleExclusiveRequiredOptions() { // present > 1 config.put(TEST_TOPIC.key(), "[\"saitou\"]"); - assertEquals("ErrorCode:[API-02], ErrorDescription:[Option item validate failed] - These options(['option.topic-pattern'], ['option.topic']) are mutually exclusive, " + + assertEquals("ErrorCode:[API-02], ErrorDescription:[Option item validate failed] - These options('option.topic-pattern', 'option.topic') are mutually exclusive, " + "allowing only one set(\"[] for a set\") of options to be configured.", assertThrows(OptionValidationException.class, executable).getMessage()); } @Test public void testComplexExclusiveRequiredOptions() { - List exclusiveBundledOptions = new ArrayList<>(); - exclusiveBundledOptions.add(RequiredOption.BundledRequiredOptions.of(KEY_USERNAME, KEY_PASSWORD)); OptionRule rule = OptionRule.builder() - .exclusive(RequiredOption.ExclusiveRequiredOptions.of(exclusiveBundledOptions, KEY_BEARER_TOKEN, KEY_KERBEROS_TICKET)) + .exclusive(KEY_BEARER_TOKEN, KEY_KERBEROS_TICKET) .build(); Map config = new HashMap<>(); Executable executable = () -> validate(config, rule); // all absent - assertEquals("ErrorCode:[API-02], ErrorDescription:[Option item validate failed] - There are unconfigured options, these options(['bearer-token'], ['kerberos-ticket'], ['username', 'password']) are mutually exclusive," + + assertEquals("ErrorCode:[API-02], ErrorDescription:[Option item validate failed] - There are unconfigured options, these options('bearer-token', 'kerberos-ticket') are mutually exclusive," + " allowing only one set(\"[] for a set\") of options to be configured.", assertThrows(OptionValidationException.class, executable).getMessage()); - // bundled option some present - config.put(KEY_USERNAME.key(), "asuka"); - assertEquals("ErrorCode:[API-02], ErrorDescription:[Option item validate failed] - These options('username', 'password') are bundled, must be present or absent together." + - " The options present are: 'username'. The options absent are 'password'.", - assertThrows(OptionValidationException.class, executable).getMessage()); - - // only one set options present - config.put(KEY_PASSWORD.key(), "saitou"); - Assertions.assertDoesNotThrow(executable); - - // tow set options present + // set one config.put(KEY_BEARER_TOKEN.key(), "ashulin"); - assertEquals("ErrorCode:[API-02], ErrorDescription:[Option item validate failed] - These options(['bearer-token'], ['username', 'password']) are mutually exclusive," + - " allowing only one set(\"[] for a set\") of options to be configured.", - assertThrows(OptionValidationException.class, executable).getMessage()); + Assertions.assertDoesNotThrow(executable); - // three set options present + // all set config.put(KEY_KERBEROS_TICKET.key(), "zongwen"); - assertEquals("ErrorCode:[API-02], ErrorDescription:[Option item validate failed] - These options(['bearer-token'], ['kerberos-ticket'], ['username', 'password']) are mutually exclusive," + + assertEquals("ErrorCode:[API-02], ErrorDescription:[Option item validate failed] - These options('bearer-token', 'kerberos-ticket') are mutually exclusive," + " allowing only one set(\"[] for a set\") of options to be configured.", assertThrows(OptionValidationException.class, executable).getMessage()); } @@ -177,6 +162,7 @@ public void testComplexExclusiveRequiredOptions() { @Test public void testSimpleConditionalRequiredOptionsWithDefaultValue() { OptionRule rule = OptionRule.builder() + .optional(TEST_MODE) .conditional(TEST_MODE, OptionTest.TestMode.TIMESTAMP, TEST_TIMESTAMP) .build(); Map config = new HashMap<>(); @@ -203,6 +189,7 @@ public void testSimpleConditionalRequiredOptionsWithDefaultValue() { @Test public void testSimpleConditionalRequiredOptionsWithoutDefaultValue() { OptionRule rule = OptionRule.builder() + .optional(KEY_USERNAME) .conditional(KEY_USERNAME, "ashulin", TEST_TIMESTAMP) .build(); Map config = new HashMap<>(); @@ -228,11 +215,9 @@ public void testSimpleConditionalRequiredOptionsWithoutDefaultValue() { @Test public void testComplexConditionalRequiredOptions() { - Expression expression = Expression.of(KEY_USERNAME, "ashulin") - .or(Expression.of(Condition.of(KEY_USERNAME, "asuka") - .and(KEY_PASSWORD, "saito"))); OptionRule rule = OptionRule.builder() - .conditional(expression, TEST_TIMESTAMP) + .optional(KEY_USERNAME) + .conditional(KEY_USERNAME, Arrays.asList("ashulin", "asuka"), TEST_TIMESTAMP) .build(); Map config = new HashMap<>(); Executable executable = () -> validate(config, rule); @@ -243,7 +228,13 @@ public void testComplexConditionalRequiredOptions() { // 'username' == ashulin, and required options absent config.put(KEY_USERNAME.key(), "ashulin"); assertEquals("ErrorCode:[API-02], ErrorDescription:[Option item validate failed] - There are unconfigured options, the options('option.timestamp') are required" + - " because ['username' == ashulin || ('username' == asuka && 'password' == saito)] is true.", + " because ['username' == ashulin || 'username' == asuka] is true.", + assertThrows(OptionValidationException.class, executable).getMessage()); + + // 'username' == asuka, and required options absent + config.put(KEY_USERNAME.key(), "asuka"); + assertEquals("ErrorCode:[API-02], ErrorDescription:[Option item validate failed] - There are unconfigured options, the options('option.timestamp') are required" + + " because ['username' == ashulin || 'username' == asuka] is true.", assertThrows(OptionValidationException.class, executable).getMessage()); // Expression match, and required options all present @@ -251,17 +242,7 @@ public void testComplexConditionalRequiredOptions() { Assertions.assertDoesNotThrow(executable); // Expression mismatch - config.put(KEY_USERNAME.key(), "asuka"); + config.put(KEY_USERNAME.key(), "asuka111"); Assertions.assertDoesNotThrow(executable); - - // 'username' == asuka && 'password' == saito - config.put(KEY_PASSWORD.key(), "saito"); - Assertions.assertDoesNotThrow(executable); - - // 'username' == asuka && 'password' == saito, and required options absent - config.remove(TEST_TIMESTAMP.key()); - assertEquals("ErrorCode:[API-02], ErrorDescription:[Option item validate failed] - There are unconfigured options, the options('option.timestamp') are required" + - " because ['username' == ashulin || ('username' == asuka && 'password' == saito)] is true.", - assertThrows(OptionValidationException.class, executable).getMessage()); } } diff --git a/seatunnel-api/src/test/java/org/apache/seatunnel/api/configuration/util/OptionRuleTest.java b/seatunnel-api/src/test/java/org/apache/seatunnel/api/configuration/util/OptionRuleTest.java index c1a5cff459e..b61da82c2c6 100644 --- a/seatunnel-api/src/test/java/org/apache/seatunnel/api/configuration/util/OptionRuleTest.java +++ b/seatunnel-api/src/test/java/org/apache/seatunnel/api/configuration/util/OptionRuleTest.java @@ -19,6 +19,8 @@ import static org.apache.seatunnel.api.configuration.OptionTest.TEST_MODE; import static org.apache.seatunnel.api.configuration.OptionTest.TEST_NUM; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; import org.apache.seatunnel.api.configuration.Option; import org.apache.seatunnel.api.configuration.OptionTest; @@ -27,6 +29,7 @@ import com.fasterxml.jackson.core.type.TypeReference; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.function.Executable; import java.util.List; @@ -52,6 +55,16 @@ public class OptionRuleTest { .noDefaultValue() .withDescription("test list int type"); + public static final Option TEST_REQUIRED_HAVE_DEFAULT_VALUE = Options.key("option.required-have-default") + .stringType() + .defaultValue("11") + .withDescription("test string type"); + + public static final Option TEST_DUPLICATE = Options.key("option.test-duplicate") + .stringType() + .noDefaultValue() + .withDescription("test string type"); + @Test public void testBuildSuccess() { OptionRule rule = OptionRule.builder() @@ -63,6 +76,80 @@ public void testBuildSuccess() { Assertions.assertNotNull(rule); } + @Test + public void testVerify() { + Executable executable = () -> { + OptionRule.builder() + .optional(TEST_NUM, TEST_MODE) + .required(TEST_PORTS, TEST_REQUIRED_HAVE_DEFAULT_VALUE) + .exclusive(TEST_TOPIC_PATTERN, TEST_TOPIC) + .conditional(TEST_MODE, OptionTest.TestMode.TIMESTAMP, TEST_TIMESTAMP) + .build(); + }; + + // test required option have no default value + assertEquals( + "ErrorCode:[API-02], ErrorDescription:[Option item validate failed] - Required option 'option.required-have-default' should have no default value.", + assertThrows(OptionValidationException.class, executable).getMessage()); + + executable = () -> { + OptionRule.builder() + .optional(TEST_NUM, TEST_MODE, TEST_REQUIRED_HAVE_DEFAULT_VALUE) + .required(TEST_PORTS, TEST_REQUIRED_HAVE_DEFAULT_VALUE) + .exclusive(TEST_TOPIC_PATTERN, TEST_TOPIC) + .conditional(TEST_MODE, OptionTest.TestMode.TIMESTAMP, TEST_TIMESTAMP) + .build(); + }; + + // test duplicate + assertEquals( + "ErrorCode:[API-02], ErrorDescription:[Option item validate failed] - RequiredOption 'option.required-have-default' duplicate in option options.", + assertThrows(OptionValidationException.class, executable).getMessage()); + + executable = () -> { + OptionRule.builder() + .optional(TEST_NUM, TEST_MODE) + .exclusive(TEST_TOPIC_PATTERN, TEST_TOPIC, TEST_DUPLICATE) + .required(TEST_PORTS, TEST_DUPLICATE) + .conditional(TEST_MODE, OptionTest.TestMode.TIMESTAMP, TEST_TIMESTAMP) + .build(); + }; + + // test duplicate in RequiredOption$ExclusiveRequiredOptions + assertEquals( + "ErrorCode:[API-02], ErrorDescription:[Option item validate failed] - RequiredOption 'option.test-duplicate' duplicate in 'org.apache.seatunnel.api.configuration.util.RequiredOption$ExclusiveRequiredOptions'.", + assertThrows(OptionValidationException.class, executable).getMessage()); + + executable = () -> { + OptionRule.builder() + .optional(TEST_NUM) + .exclusive(TEST_TOPIC_PATTERN, TEST_TOPIC) + .required(TEST_PORTS) + .conditional(TEST_MODE, OptionTest.TestMode.TIMESTAMP, TEST_TIMESTAMP) + .build(); + }; + + // test conditional not found in other options + assertEquals( + "ErrorCode:[API-02], ErrorDescription:[Option item validate failed] - Conditional 'option.mode' not found in options.", + assertThrows(OptionValidationException.class, executable).getMessage()); + + executable = () -> { + OptionRule.builder() + .optional(TEST_NUM, TEST_MODE) + .exclusive(TEST_TOPIC_PATTERN, TEST_TOPIC) + .required(TEST_PORTS) + .conditional(TEST_MODE, OptionTest.TestMode.TIMESTAMP, TEST_TIMESTAMP) + .conditional(TEST_NUM, 100, TEST_TIMESTAMP) + .build(); + }; + + // test parameter can only be controlled by one other parameter + assertEquals( + "ErrorCode:[API-02], ErrorDescription:[Option item validate failed] - ConditionalOption 'option.timestamp' duplicate in 'org.apache.seatunnel.api.configuration.util.RequiredOption$ConditionalRequiredOptions'.", + assertThrows(OptionValidationException.class, executable).getMessage()); + } + @Test public void testEquals() { OptionRule rule1 = OptionRule.builder() diff --git a/seatunnel-api/src/test/java/org/apache/seatunnel/api/configuration/util/ExpressionTest.java b/seatunnel-connectors-v2/connector-assert/src/test/java/org/apache/seatunnel/flink/assertion/AssertFactoryTest.java similarity index 54% rename from seatunnel-api/src/test/java/org/apache/seatunnel/api/configuration/util/ExpressionTest.java rename to seatunnel-connectors-v2/connector-assert/src/test/java/org/apache/seatunnel/flink/assertion/AssertFactoryTest.java index e7971a1a3f2..cf9473b5f19 100644 --- a/seatunnel-api/src/test/java/org/apache/seatunnel/api/configuration/util/ExpressionTest.java +++ b/seatunnel-connectors-v2/connector-assert/src/test/java/org/apache/seatunnel/flink/assertion/AssertFactoryTest.java @@ -15,23 +15,20 @@ * limitations under the License. */ -package org.apache.seatunnel.api.configuration.util; +package org.apache.seatunnel.flink.assertion; -import static org.apache.seatunnel.api.configuration.OptionTest.TEST_MODE; -import static org.apache.seatunnel.api.configuration.OptionTest.TEST_NUM; - -import org.apache.seatunnel.api.configuration.OptionTest; +import org.apache.seatunnel.api.configuration.util.OptionRule; +import org.apache.seatunnel.connectors.seatunnel.assertion.sink.AssertSinkFactory; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; -public class ExpressionTest { +public class AssertFactoryTest { + @Test - public void testToString() { - Expression expression = Expression.of(TEST_NUM, 200) - .and(Expression.of(Condition.of(TEST_MODE, OptionTest.TestMode.EARLIEST) - .or(TEST_MODE, OptionTest.TestMode.LATEST))) - .or(Expression.of(Condition.of(TEST_NUM, 100))); - Assertions.assertEquals("('option.num' == 200 && ('option.mode' == EARLIEST || 'option.mode' == LATEST)) || 'option.num' == 100", expression.toString()); + public void testOptionRule() throws Exception { + AssertSinkFactory factory = new AssertSinkFactory(); + OptionRule optionRule = factory.optionRule(); + Assertions.assertNotNull(optionRule); } } diff --git a/seatunnel-connectors-v2/connector-clickhouse/src/test/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/ClickhouseFactoryTest.java b/seatunnel-connectors-v2/connector-clickhouse/src/test/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/ClickhouseFactoryTest.java new file mode 100644 index 00000000000..2a4205016ed --- /dev/null +++ b/seatunnel-connectors-v2/connector-clickhouse/src/test/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/ClickhouseFactoryTest.java @@ -0,0 +1,33 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.clickhouse; + +import org.apache.seatunnel.connectors.seatunnel.clickhouse.sink.ClickhouseSinkFactory; +import org.apache.seatunnel.connectors.seatunnel.clickhouse.source.ClickhouseSourceFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +public class ClickhouseFactoryTest { + + @Test + public void testOptionRule() { + Assertions.assertNotNull((new ClickhouseSourceFactory()).optionRule()); + Assertions.assertNotNull((new ClickhouseSinkFactory()).optionRule()); + } +} diff --git a/seatunnel-connectors-v2/connector-console/src/test/java/org/apache/seatunnel/connectors/seatunnel/console/ConsoleFactoryTest.java b/seatunnel-connectors-v2/connector-console/src/test/java/org/apache/seatunnel/connectors/seatunnel/console/ConsoleFactoryTest.java new file mode 100644 index 00000000000..59ade0f0ffc --- /dev/null +++ b/seatunnel-connectors-v2/connector-console/src/test/java/org/apache/seatunnel/connectors/seatunnel/console/ConsoleFactoryTest.java @@ -0,0 +1,30 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.console; + +import org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +public class ConsoleFactoryTest { + @Test + public void testOptionRule() { + Assertions.assertNotNull((new ConsoleSinkFactory()).optionRule()); + } +} diff --git a/seatunnel-connectors-v2/connector-datahub/src/test/java/org/apache/seatunnel/connectors/seatunnel/datahub/DataHubFactoryTest.java b/seatunnel-connectors-v2/connector-datahub/src/test/java/org/apache/seatunnel/connectors/seatunnel/datahub/DataHubFactoryTest.java new file mode 100644 index 00000000000..3ea9a5cb694 --- /dev/null +++ b/seatunnel-connectors-v2/connector-datahub/src/test/java/org/apache/seatunnel/connectors/seatunnel/datahub/DataHubFactoryTest.java @@ -0,0 +1,31 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.datahub; + +import org.apache.seatunnel.connectors.seatunnel.datahub.sink.DataHubSinkFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +public class DataHubFactoryTest { + + @Test + void optionRule() { + Assertions.assertNotNull((new DataHubSinkFactory()).optionRule()); + } +} diff --git a/seatunnel-connectors-v2/connector-dingtalk/src/test/java/org/apache/seatunnel/connectors/seatunnel/DingTalkFactoryTest.java b/seatunnel-connectors-v2/connector-dingtalk/src/test/java/org/apache/seatunnel/connectors/seatunnel/DingTalkFactoryTest.java new file mode 100644 index 00000000000..66563cf9d53 --- /dev/null +++ b/seatunnel-connectors-v2/connector-dingtalk/src/test/java/org/apache/seatunnel/connectors/seatunnel/DingTalkFactoryTest.java @@ -0,0 +1,31 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel; + +import org.apache.seatunnel.connectors.seatunnel.sink.DingTalkSinkFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +public class DingTalkFactoryTest { + + @Test + void optionRule() { + Assertions.assertNotNull((new DingTalkSinkFactory()).optionRule()); + } +} diff --git a/seatunnel-connectors-v2/connector-elasticsearch/src/test/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/ElasticsearchFactoryTest.java b/seatunnel-connectors-v2/connector-elasticsearch/src/test/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/ElasticsearchFactoryTest.java new file mode 100644 index 00000000000..bd323e492c7 --- /dev/null +++ b/seatunnel-connectors-v2/connector-elasticsearch/src/test/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/ElasticsearchFactoryTest.java @@ -0,0 +1,33 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.elasticsearch; + +import org.apache.seatunnel.connectors.seatunnel.elasticsearch.sink.ElasticsearchSinkFactory; +import org.apache.seatunnel.connectors.seatunnel.elasticsearch.source.ElasticsearchSourceFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +public class ElasticsearchFactoryTest { + + @Test + void optionRule() { + Assertions.assertNotNull((new ElasticsearchSourceFactory()).optionRule()); + Assertions.assertNotNull((new ElasticsearchSinkFactory()).optionRule()); + } +} diff --git a/seatunnel-connectors-v2/connector-email/src/test/java/org/apache/seatunnel/connectors/seatunnel/email/EmailFactoryTest.java b/seatunnel-connectors-v2/connector-email/src/test/java/org/apache/seatunnel/connectors/seatunnel/email/EmailFactoryTest.java new file mode 100644 index 00000000000..f2f938b69dc --- /dev/null +++ b/seatunnel-connectors-v2/connector-email/src/test/java/org/apache/seatunnel/connectors/seatunnel/email/EmailFactoryTest.java @@ -0,0 +1,31 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.email; + +import org.apache.seatunnel.connectors.seatunnel.email.sink.EmailSinkFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +public class EmailFactoryTest { + + @Test + void optionRule() { + Assertions.assertNotNull((new EmailSinkFactory()).optionRule()); + } +} diff --git a/seatunnel-connectors-v2/connector-fake/src/test/java/org/apache/seatunnel/connectors/seatunnel/fake/FakeFactoryTest.java b/seatunnel-connectors-v2/connector-fake/src/test/java/org/apache/seatunnel/connectors/seatunnel/fake/FakeFactoryTest.java new file mode 100644 index 00000000000..175fc623080 --- /dev/null +++ b/seatunnel-connectors-v2/connector-fake/src/test/java/org/apache/seatunnel/connectors/seatunnel/fake/FakeFactoryTest.java @@ -0,0 +1,31 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.fake; + +import org.apache.seatunnel.connectors.seatunnel.fake.source.FakeSourceFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +public class FakeFactoryTest { + + @Test + void optionRule() { + Assertions.assertNotNull((new FakeSourceFactory()).optionRule()); + } +} diff --git a/seatunnel-connectors-v2/connector-file/connector-file-ftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/source/FtpFileSourceFactory.java b/seatunnel-connectors-v2/connector-file/connector-file-ftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/source/FtpFileSourceFactory.java index eb01aa1e0dc..300167c2492 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-ftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/source/FtpFileSourceFactory.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-ftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/source/FtpFileSourceFactory.java @@ -17,7 +17,6 @@ package org.apache.seatunnel.connectors.seatunnel.file.ftp.source; -import org.apache.seatunnel.api.configuration.util.Condition; import org.apache.seatunnel.api.configuration.util.OptionRule; import org.apache.seatunnel.api.table.factory.Factory; import org.apache.seatunnel.api.table.factory.TableSourceFactory; @@ -27,6 +26,8 @@ import com.google.auto.service.AutoService; +import java.util.Arrays; + @AutoService(Factory.class) public class FtpFileSourceFactory implements TableSourceFactory { @Override @@ -48,8 +49,7 @@ public OptionRule optionRule() { .optional(FtpConfig.DATE_FORMAT) .optional(FtpConfig.DATETIME_FORMAT) .optional(FtpConfig.TIME_FORMAT) - .conditional(Condition.of(FtpConfig.FILE_TYPE, "text"), SeaTunnelSchema.SCHEMA) - .conditional(Condition.of(FtpConfig.FILE_TYPE, "json"), SeaTunnelSchema.SCHEMA) + .conditional(FtpConfig.FILE_TYPE, Arrays.asList("text", "json"), SeaTunnelSchema.SCHEMA) .build(); } } diff --git a/seatunnel-connectors-v2/connector-file/connector-file-ftp/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/FtpFileFactoryTest.java b/seatunnel-connectors-v2/connector-file/connector-file-ftp/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/FtpFileFactoryTest.java new file mode 100644 index 00000000000..529d1be814b --- /dev/null +++ b/seatunnel-connectors-v2/connector-file/connector-file-ftp/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/FtpFileFactoryTest.java @@ -0,0 +1,33 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.file.ftp; + +import org.apache.seatunnel.connectors.seatunnel.file.ftp.sink.FtpFileSinkFactory; +import org.apache.seatunnel.connectors.seatunnel.file.ftp.source.FtpFileSourceFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class FtpFileFactoryTest { + + @Test + void optionRule() { + Assertions.assertNotNull((new FtpFileSourceFactory()).optionRule()); + Assertions.assertNotNull((new FtpFileSinkFactory()).optionRule()); + } +} diff --git a/seatunnel-connectors-v2/connector-file/connector-file-hadoop/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hdfs/source/HdfsFileSourceFactory.java b/seatunnel-connectors-v2/connector-file/connector-file-hadoop/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hdfs/source/HdfsFileSourceFactory.java index e1bf3dd3f6d..40b3a807c13 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-hadoop/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hdfs/source/HdfsFileSourceFactory.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-hadoop/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hdfs/source/HdfsFileSourceFactory.java @@ -17,7 +17,6 @@ package org.apache.seatunnel.connectors.seatunnel.file.hdfs.source; -import org.apache.seatunnel.api.configuration.util.Condition; import org.apache.seatunnel.api.configuration.util.OptionRule; import org.apache.seatunnel.api.table.factory.Factory; import org.apache.seatunnel.api.table.factory.TableSourceFactory; @@ -27,6 +26,8 @@ import com.google.auto.service.AutoService; +import java.util.Arrays; + @AutoService(Factory.class) public class HdfsFileSourceFactory implements TableSourceFactory { @Override @@ -45,8 +46,7 @@ public OptionRule optionRule() { .optional(HdfsSourceConfig.DATE_FORMAT) .optional(HdfsSourceConfig.DATETIME_FORMAT) .optional(HdfsSourceConfig.TIME_FORMAT) - .conditional(Condition.of(HdfsSourceConfig.FILE_TYPE, "text"), SeaTunnelSchema.SCHEMA) - .conditional(Condition.of(HdfsSourceConfig.FILE_TYPE, "json"), SeaTunnelSchema.SCHEMA) + .conditional(HdfsSourceConfig.FILE_TYPE, Arrays.asList("text", "json"), SeaTunnelSchema.SCHEMA) .build(); } } diff --git a/seatunnel-connectors-v2/connector-file/connector-file-hadoop/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/hdfs/HdfsFileFactoryTest.java b/seatunnel-connectors-v2/connector-file/connector-file-hadoop/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/hdfs/HdfsFileFactoryTest.java new file mode 100644 index 00000000000..1a9171dedcb --- /dev/null +++ b/seatunnel-connectors-v2/connector-file/connector-file-hadoop/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/hdfs/HdfsFileFactoryTest.java @@ -0,0 +1,33 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.file.hdfs; + +import org.apache.seatunnel.connectors.seatunnel.file.hdfs.sink.HdfsFileSinkFactory; +import org.apache.seatunnel.connectors.seatunnel.file.hdfs.source.HdfsFileSourceFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class HdfsFileFactoryTest { + + @Test + void optionRule() { + Assertions.assertNotNull((new HdfsFileSourceFactory()).optionRule()); + Assertions.assertNotNull((new HdfsFileSinkFactory()).optionRule()); + } +} diff --git a/seatunnel-connectors-v2/connector-file/connector-file-local/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/local/source/LocalFileSourceFactory.java b/seatunnel-connectors-v2/connector-file/connector-file-local/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/local/source/LocalFileSourceFactory.java index c9cfad1654e..bf02203944d 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-local/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/local/source/LocalFileSourceFactory.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-local/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/local/source/LocalFileSourceFactory.java @@ -17,7 +17,6 @@ package org.apache.seatunnel.connectors.seatunnel.file.local.source; -import org.apache.seatunnel.api.configuration.util.Condition; import org.apache.seatunnel.api.configuration.util.OptionRule; import org.apache.seatunnel.api.table.factory.Factory; import org.apache.seatunnel.api.table.factory.TableSourceFactory; @@ -27,6 +26,8 @@ import com.google.auto.service.AutoService; +import java.util.Arrays; + @AutoService(Factory.class) public class LocalFileSourceFactory implements TableSourceFactory { @Override @@ -44,8 +45,7 @@ public OptionRule optionRule() { .optional(LocalSourceConfig.DATE_FORMAT) .optional(LocalSourceConfig.DATETIME_FORMAT) .optional(LocalSourceConfig.TIME_FORMAT) - .conditional(Condition.of(LocalSourceConfig.FILE_TYPE, "text"), SeaTunnelSchema.SCHEMA) - .conditional(Condition.of(LocalSourceConfig.FILE_TYPE, "json"), SeaTunnelSchema.SCHEMA) + .conditional(LocalSourceConfig.FILE_TYPE, Arrays.asList("text", "json"), SeaTunnelSchema.SCHEMA) .build(); } } diff --git a/seatunnel-connectors-v2/connector-file/connector-file-local/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/local/LocalFileFactoryTest.java b/seatunnel-connectors-v2/connector-file/connector-file-local/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/local/LocalFileFactoryTest.java new file mode 100644 index 00000000000..693f81578a1 --- /dev/null +++ b/seatunnel-connectors-v2/connector-file/connector-file-local/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/local/LocalFileFactoryTest.java @@ -0,0 +1,33 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.file.local; + +import org.apache.seatunnel.connectors.seatunnel.file.local.sink.LocalFileSinkFactory; +import org.apache.seatunnel.connectors.seatunnel.file.local.source.LocalFileSourceFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class LocalFileFactoryTest { + + @Test + void optionRule() { + Assertions.assertNotNull((new LocalFileSinkFactory()).optionRule()); + Assertions.assertNotNull((new LocalFileSourceFactory()).optionRule()); + } +} diff --git a/seatunnel-connectors-v2/connector-file/connector-file-oss/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/oss/source/OssFileSourceFactory.java b/seatunnel-connectors-v2/connector-file/connector-file-oss/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/oss/source/OssFileSourceFactory.java index cd875c0e841..f34d2aadcb9 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-oss/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/oss/source/OssFileSourceFactory.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-oss/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/oss/source/OssFileSourceFactory.java @@ -17,7 +17,6 @@ package org.apache.seatunnel.connectors.seatunnel.file.oss.source; -import org.apache.seatunnel.api.configuration.util.Condition; import org.apache.seatunnel.api.configuration.util.OptionRule; import org.apache.seatunnel.api.table.factory.Factory; import org.apache.seatunnel.api.table.factory.TableSourceFactory; @@ -27,6 +26,8 @@ import com.google.auto.service.AutoService; +import java.util.Arrays; + @AutoService(Factory.class) public class OssFileSourceFactory implements TableSourceFactory { @Override @@ -48,8 +49,7 @@ public OptionRule optionRule() { .optional(OssConfig.DATE_FORMAT) .optional(OssConfig.DATETIME_FORMAT) .optional(OssConfig.TIME_FORMAT) - .conditional(Condition.of(OssConfig.FILE_TYPE, "text"), SeaTunnelSchema.SCHEMA) - .conditional(Condition.of(OssConfig.FILE_TYPE, "json"), SeaTunnelSchema.SCHEMA) + .conditional(OssConfig.FILE_TYPE, Arrays.asList("text", "json"), SeaTunnelSchema.SCHEMA) .build(); } } diff --git a/seatunnel-connectors-v2/connector-file/connector-file-oss/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/oss/OssFileFactoryTest.java b/seatunnel-connectors-v2/connector-file/connector-file-oss/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/oss/OssFileFactoryTest.java new file mode 100644 index 00000000000..fe7b37c4d20 --- /dev/null +++ b/seatunnel-connectors-v2/connector-file/connector-file-oss/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/oss/OssFileFactoryTest.java @@ -0,0 +1,33 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.file.oss; + +import org.apache.seatunnel.connectors.seatunnel.file.oss.sink.OssFileSinkFactory; +import org.apache.seatunnel.connectors.seatunnel.file.oss.source.OssFileSourceFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +public class OssFileFactoryTest { + + @Test + void optionRule() { + Assertions.assertNotNull((new OssFileSourceFactory()).optionRule()); + Assertions.assertNotNull((new OssFileSinkFactory()).optionRule()); + } +} diff --git a/seatunnel-connectors-v2/connector-file/connector-file-s3/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/s3/source/S3FileSourceFactory.java b/seatunnel-connectors-v2/connector-file/connector-file-s3/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/s3/source/S3FileSourceFactory.java index 018534e6266..b9985690786 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-s3/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/s3/source/S3FileSourceFactory.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-s3/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/s3/source/S3FileSourceFactory.java @@ -17,7 +17,6 @@ package org.apache.seatunnel.connectors.seatunnel.file.s3.source; -import org.apache.seatunnel.api.configuration.util.Condition; import org.apache.seatunnel.api.configuration.util.OptionRule; import org.apache.seatunnel.api.table.factory.Factory; import org.apache.seatunnel.api.table.factory.TableSourceFactory; @@ -27,6 +26,8 @@ import com.google.auto.service.AutoService; +import java.util.Arrays; + @AutoService(Factory.class) public class S3FileSourceFactory implements TableSourceFactory { @Override @@ -47,8 +48,7 @@ public OptionRule optionRule() { .optional(S3Config.DATE_FORMAT) .optional(S3Config.DATETIME_FORMAT) .optional(S3Config.TIME_FORMAT) - .conditional(Condition.of(S3Config.FILE_TYPE, "text"), SeaTunnelSchema.SCHEMA) - .conditional(Condition.of(S3Config.FILE_TYPE, "json"), SeaTunnelSchema.SCHEMA) + .conditional(S3Config.FILE_TYPE, Arrays.asList("text", "json"), SeaTunnelSchema.SCHEMA) .build(); } } diff --git a/seatunnel-connectors-v2/connector-file/connector-file-s3/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/s3/S3FileFactoryTest.java b/seatunnel-connectors-v2/connector-file/connector-file-s3/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/s3/S3FileFactoryTest.java new file mode 100644 index 00000000000..f52a16e2641 --- /dev/null +++ b/seatunnel-connectors-v2/connector-file/connector-file-s3/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/s3/S3FileFactoryTest.java @@ -0,0 +1,33 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.file.s3; + +import org.apache.seatunnel.connectors.seatunnel.file.s3.sink.S3FileSinkFactory; +import org.apache.seatunnel.connectors.seatunnel.file.s3.source.S3FileSourceFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class S3FileFactoryTest { + + @Test + void optionRule() { + Assertions.assertNotNull((new S3FileSourceFactory()).optionRule()); + Assertions.assertNotNull((new S3FileSinkFactory()).optionRule()); + } +} diff --git a/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/source/SftpFileSourceFactory.java b/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/source/SftpFileSourceFactory.java index 448f7a67a77..bf6e5b382ad 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/source/SftpFileSourceFactory.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/source/SftpFileSourceFactory.java @@ -17,7 +17,6 @@ package org.apache.seatunnel.connectors.seatunnel.file.sftp.source; -import org.apache.seatunnel.api.configuration.util.Condition; import org.apache.seatunnel.api.configuration.util.OptionRule; import org.apache.seatunnel.api.table.factory.Factory; import org.apache.seatunnel.api.table.factory.TableSourceFactory; @@ -27,6 +26,8 @@ import com.google.auto.service.AutoService; +import java.util.Arrays; + @AutoService(Factory.class) public class SftpFileSourceFactory implements TableSourceFactory { @Override @@ -48,8 +49,7 @@ public OptionRule optionRule() { .optional(SftpConfig.DATE_FORMAT) .optional(SftpConfig.DATETIME_FORMAT) .optional(SftpConfig.TIME_FORMAT) - .conditional(Condition.of(SftpConfig.FILE_TYPE, "text"), SeaTunnelSchema.SCHEMA) - .conditional(Condition.of(SftpConfig.FILE_TYPE, "json"), SeaTunnelSchema.SCHEMA) + .conditional(SftpConfig.FILE_TYPE, Arrays.asList("text", "json"), SeaTunnelSchema.SCHEMA) .build(); } } diff --git a/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/SftpFileFactoryTest.java b/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/SftpFileFactoryTest.java new file mode 100644 index 00000000000..d1964d3e98a --- /dev/null +++ b/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/SftpFileFactoryTest.java @@ -0,0 +1,33 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.file.sftp; + +import org.apache.seatunnel.connectors.seatunnel.file.sftp.sink.SftpFileSinkFactory; +import org.apache.seatunnel.connectors.seatunnel.file.sftp.source.SftpFileSourceFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class SftpFileFactoryTest { + + @Test + void optionRule() { + Assertions.assertNotNull((new SftpFileSourceFactory()).optionRule()); + Assertions.assertNotNull((new SftpFileSinkFactory()).optionRule()); + } +} diff --git a/seatunnel-connectors-v2/connector-google-sheets/src/test/java/org/apache/seatunnel/connectors/seatunnel/google/sheets/SheetsFactoryTest.java b/seatunnel-connectors-v2/connector-google-sheets/src/test/java/org/apache/seatunnel/connectors/seatunnel/google/sheets/SheetsFactoryTest.java new file mode 100644 index 00000000000..3144a403dc2 --- /dev/null +++ b/seatunnel-connectors-v2/connector-google-sheets/src/test/java/org/apache/seatunnel/connectors/seatunnel/google/sheets/SheetsFactoryTest.java @@ -0,0 +1,31 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.google.sheets; + +import org.apache.seatunnel.connectors.seatunnel.google.sheets.source.SheetsSourceFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class SheetsFactoryTest { + + @Test + void optionRule() { + Assertions.assertNotNull((new SheetsSourceFactory()).optionRule()); + } +} diff --git a/seatunnel-connectors-v2/connector-hive/src/test/java/org/apache/seatunnel/connectors/seatunnel/hive/HiveFactoryTest.java b/seatunnel-connectors-v2/connector-hive/src/test/java/org/apache/seatunnel/connectors/seatunnel/hive/HiveFactoryTest.java new file mode 100644 index 00000000000..7909fd24116 --- /dev/null +++ b/seatunnel-connectors-v2/connector-hive/src/test/java/org/apache/seatunnel/connectors/seatunnel/hive/HiveFactoryTest.java @@ -0,0 +1,33 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.hive; + +import org.apache.seatunnel.connectors.seatunnel.hive.sink.HiveSinkFactory; +import org.apache.seatunnel.connectors.seatunnel.hive.source.HiveSourceFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class HiveFactoryTest { + + @Test + void optionRule() { + Assertions.assertNotNull((new HiveSourceFactory()).optionRule()); + Assertions.assertNotNull((new HiveSinkFactory()).optionRule()); + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/config/HttpConfig.java b/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/config/HttpConfig.java index 1e605f1fade..2a2e4058c0d 100644 --- a/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/config/HttpConfig.java +++ b/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/config/HttpConfig.java @@ -23,48 +23,66 @@ import java.util.Map; public class HttpConfig { - public static final String DEFAULT_FORMAT = "json"; public static final String BASIC = "Basic"; public static final int DEFAULT_RETRY_BACKOFF_MULTIPLIER_MS = 100; public static final int DEFAULT_RETRY_BACKOFF_MAX_MS = 10000; public static final Option URL = Options.key("url") - .stringType() - .noDefaultValue() - .withDescription("Http request url"); + .stringType() + .noDefaultValue() + .withDescription("Http request url"); public static final Option METHOD = Options.key("method") - .enumType(HttpRequestMethod.class) - .defaultValue(HttpRequestMethod.GET) - .withDescription("Http request method"); + .enumType(HttpRequestMethod.class) + .defaultValue(HttpRequestMethod.GET) + .withDescription("Http request method"); public static final Option> HEADERS = Options.key("headers") - .mapType() - .noDefaultValue() - .withDescription("Http request headers"); + .mapType() + .noDefaultValue() + .withDescription("Http request headers"); public static final Option> PARAMS = Options.key("params") - .mapType() - .noDefaultValue() - .withDescription("Http request params"); + .mapType() + .noDefaultValue() + .withDescription("Http request params"); public static final Option BODY = Options.key("body") - .stringType() - .noDefaultValue() - .withDescription("Http request body"); - public static final Option FORMAT = Options.key("format") - .stringType() - .defaultValue(DEFAULT_FORMAT) - .withDescription("Http response format"); + .stringType() + .noDefaultValue() + .withDescription("Http request body"); + public static final Option FORMAT = Options.key("format") + .enumType(ResponseFormat.class) + .defaultValue(ResponseFormat.JSON) + .withDescription("Http response format"); public static final Option POLL_INTERVAL_MILLS = Options.key("poll_interval_millis") - .intType() - .noDefaultValue() - .withDescription("Request http api interval(millis) in stream mode"); + .intType() + .noDefaultValue() + .withDescription("Request http api interval(millis) in stream mode"); public static final Option RETRY = Options.key("retry") - .intType() - .noDefaultValue() - .withDescription("The max retry times if request http return to IOException"); + .intType() + .noDefaultValue() + .withDescription("The max retry times if request http return to IOException"); public static final Option RETRY_BACKOFF_MULTIPLIER_MS = Options.key("retry_backoff_multiplier_ms") - .intType() - .defaultValue(DEFAULT_RETRY_BACKOFF_MULTIPLIER_MS) - .withDescription("The retry-backoff times(millis) multiplier if request http failed"); + .intType() + .defaultValue(DEFAULT_RETRY_BACKOFF_MULTIPLIER_MS) + .withDescription("The retry-backoff times(millis) multiplier if request http failed"); public static final Option RETRY_BACKOFF_MAX_MS = Options.key("retry_backoff_max_ms") - .intType() - .defaultValue(DEFAULT_RETRY_BACKOFF_MAX_MS) - .withDescription("The maximum retry-backoff times(millis) if request http failed"); + .intType() + .defaultValue(DEFAULT_RETRY_BACKOFF_MAX_MS) + .withDescription("The maximum retry-backoff times(millis) if request http failed"); + + public enum ResponseFormat { + JSON("json"); + + private String format; + + ResponseFormat(String format) { + this.format = format; + } + + public String getFormat() { + return format; + } + + @Override + public String toString() { + return format; + } + } } diff --git a/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/source/HttpSource.java b/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/source/HttpSource.java index 69759f16c43..3f050c478e5 100644 --- a/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/source/HttpSource.java +++ b/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/source/HttpSource.java @@ -41,6 +41,8 @@ import com.google.auto.service.AutoService; +import java.util.Locale; + @AutoService(SeaTunnelSource.class) public class HttpSource extends AbstractSingleSplitSource { protected final HttpParameter httpParameter = new HttpParameter(); @@ -73,12 +75,13 @@ protected void buildSchemaWithConfig(Config pluginConfig) { Config schema = pluginConfig.getConfig(SeaTunnelSchema.SCHEMA.key()); this.rowType = SeaTunnelSchema.buildWithConfig(schema).getSeaTunnelRowType(); // default use json format - String format = HttpConfig.DEFAULT_FORMAT; + HttpConfig.ResponseFormat format = HttpConfig.FORMAT.defaultValue(); if (pluginConfig.hasPath(HttpConfig.FORMAT.key())) { - format = pluginConfig.getString(HttpConfig.FORMAT.key()); + format = HttpConfig.ResponseFormat.valueOf(pluginConfig.getString(HttpConfig.FORMAT.key()).toUpperCase( + Locale.ROOT)); } switch (format) { - case HttpConfig.DEFAULT_FORMAT: + case JSON: this.deserializationSchema = new JsonDeserializationSchema(false, false, rowType); break; default: @@ -102,7 +105,8 @@ public SeaTunnelDataType getProducedType() { } @Override - public AbstractSingleSplitReader createReader(SingleSplitReaderContext readerContext) throws Exception { + public AbstractSingleSplitReader createReader(SingleSplitReaderContext readerContext) + throws Exception { return new HttpSourceReader(this.httpParameter, readerContext, this.deserializationSchema); } } diff --git a/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/source/HttpSourceFactory.java b/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/source/HttpSourceFactory.java index 5b7eb0c6fd6..49a5f1eb3c8 100644 --- a/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/source/HttpSourceFactory.java +++ b/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/source/HttpSourceFactory.java @@ -17,7 +17,6 @@ package org.apache.seatunnel.connectors.seatunnel.http.source; -import org.apache.seatunnel.api.configuration.util.Condition; import org.apache.seatunnel.api.configuration.util.OptionRule; import org.apache.seatunnel.api.table.factory.Factory; import org.apache.seatunnel.api.table.factory.TableSourceFactory; @@ -38,17 +37,17 @@ public String factoryIdentifier() { @Override public OptionRule optionRule() { return OptionRule.builder() - .required(HttpConfig.URL) - .optional(HttpConfig.METHOD) - .optional(HttpConfig.HEADERS) - .optional(HttpConfig.PARAMS) - .conditional(Condition.of(HttpConfig.METHOD, HttpRequestMethod.POST), HttpConfig.BODY) - .conditional(Condition.of(HttpConfig.FORMAT, "json"), SeaTunnelSchema.SCHEMA) - .optional(HttpConfig.FORMAT) - .optional(HttpConfig.POLL_INTERVAL_MILLS) - .optional(HttpConfig.RETRY) - .optional(HttpConfig.RETRY_BACKOFF_MULTIPLIER_MS) - .optional(HttpConfig.RETRY_BACKOFF_MAX_MS) - .build(); + .required(HttpConfig.URL) + .optional(HttpConfig.METHOD) + .optional(HttpConfig.HEADERS) + .optional(HttpConfig.PARAMS) + .optional(HttpConfig.FORMAT) + .conditional(HttpConfig.METHOD, HttpRequestMethod.POST, HttpConfig.BODY) + .conditional(HttpConfig.FORMAT, HttpConfig.ResponseFormat.JSON, SeaTunnelSchema.SCHEMA) + .optional(HttpConfig.POLL_INTERVAL_MILLS) + .optional(HttpConfig.RETRY) + .optional(HttpConfig.RETRY_BACKOFF_MULTIPLIER_MS) + .optional(HttpConfig.RETRY_BACKOFF_MAX_MS) + .build(); } } diff --git a/seatunnel-connectors-v2/connector-http/connector-http-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/http/HttpFactoryTest.java b/seatunnel-connectors-v2/connector-http/connector-http-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/http/HttpFactoryTest.java new file mode 100644 index 00000000000..44b73c87575 --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/http/HttpFactoryTest.java @@ -0,0 +1,33 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.http; + +import org.apache.seatunnel.connectors.seatunnel.http.sink.HttpSinkFactory; +import org.apache.seatunnel.connectors.seatunnel.http.source.HttpSourceFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class HttpFactoryTest { + + @Test + void optionRule() { + Assertions.assertNotNull((new HttpSourceFactory()).optionRule()); + Assertions.assertNotNull((new HttpSinkFactory()).optionRule()); + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-gitlab/src/main/java/org/apache/seatunnel/connectors/seatunnel/gitlab/source/GitlabSourceFactory.java b/seatunnel-connectors-v2/connector-http/connector-http-gitlab/src/main/java/org/apache/seatunnel/connectors/seatunnel/gitlab/source/GitlabSourceFactory.java index 3f1d33b71e1..76423de5d0c 100644 --- a/seatunnel-connectors-v2/connector-http/connector-http-gitlab/src/main/java/org/apache/seatunnel/connectors/seatunnel/gitlab/source/GitlabSourceFactory.java +++ b/seatunnel-connectors-v2/connector-http/connector-http-gitlab/src/main/java/org/apache/seatunnel/connectors/seatunnel/gitlab/source/GitlabSourceFactory.java @@ -17,7 +17,6 @@ package org.apache.seatunnel.connectors.seatunnel.gitlab.source; -import org.apache.seatunnel.api.configuration.util.Condition; import org.apache.seatunnel.api.configuration.util.OptionRule; import org.apache.seatunnel.api.table.factory.Factory; import org.apache.seatunnel.api.table.factory.TableSourceFactory; @@ -43,9 +42,9 @@ public OptionRule optionRule() { .optional(GitlabSourceConfig.METHOD) .optional(GitlabSourceConfig.HEADERS) .optional(GitlabSourceConfig.PARAMS) - .conditional(Condition.of(HttpConfig.METHOD, HttpRequestMethod.POST), GitlabSourceConfig.BODY) - .conditional(Condition.of(HttpConfig.FORMAT, "json"), SeaTunnelSchema.SCHEMA) .optional(GitlabSourceConfig.FORMAT) + .conditional(HttpConfig.METHOD, HttpRequestMethod.POST, GitlabSourceConfig.BODY) + .conditional(HttpConfig.FORMAT, HttpConfig.ResponseFormat.JSON, SeaTunnelSchema.SCHEMA) .optional(GitlabSourceConfig.POLL_INTERVAL_MILLS) .optional(GitlabSourceConfig.RETRY) .optional(GitlabSourceConfig.RETRY_BACKOFF_MAX_MS) diff --git a/seatunnel-connectors-v2/connector-http/connector-http-gitlab/src/test/java/org/apache/seatunnel/connectors/seatunnel/gitlab/GitlabFactoryTest.java b/seatunnel-connectors-v2/connector-http/connector-http-gitlab/src/test/java/org/apache/seatunnel/connectors/seatunnel/gitlab/GitlabFactoryTest.java new file mode 100644 index 00000000000..804873f3101 --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-gitlab/src/test/java/org/apache/seatunnel/connectors/seatunnel/gitlab/GitlabFactoryTest.java @@ -0,0 +1,31 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.gitlab; + +import org.apache.seatunnel.connectors.seatunnel.gitlab.source.GitlabSourceFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class GitlabFactoryTest { + + @Test + void optionRule() { + Assertions.assertNotNull((new GitlabSourceFactory()).optionRule()); + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-jira/src/main/java/org/apache/seatunnel/connectors/seatunnel/jira/source/JiraSourceFactory.java b/seatunnel-connectors-v2/connector-http/connector-http-jira/src/main/java/org/apache/seatunnel/connectors/seatunnel/jira/source/JiraSourceFactory.java index c37f949b5d3..e623c989cbe 100644 --- a/seatunnel-connectors-v2/connector-http/connector-http-jira/src/main/java/org/apache/seatunnel/connectors/seatunnel/jira/source/JiraSourceFactory.java +++ b/seatunnel-connectors-v2/connector-http/connector-http-jira/src/main/java/org/apache/seatunnel/connectors/seatunnel/jira/source/JiraSourceFactory.java @@ -17,7 +17,6 @@ package org.apache.seatunnel.connectors.seatunnel.jira.source; -import org.apache.seatunnel.api.configuration.util.Condition; import org.apache.seatunnel.api.configuration.util.OptionRule; import org.apache.seatunnel.api.table.factory.Factory; import org.apache.seatunnel.api.table.factory.TableSourceFactory; @@ -44,9 +43,9 @@ public OptionRule optionRule() { .optional(JiraSourceConfig.METHOD) .optional(JiraSourceConfig.HEADERS) .optional(JiraSourceConfig.PARAMS) - .conditional(Condition.of(HttpConfig.METHOD, HttpRequestMethod.POST), JiraSourceConfig.BODY) - .conditional(Condition.of(HttpConfig.FORMAT, "json"), SeaTunnelSchema.SCHEMA) .optional(JiraSourceConfig.FORMAT) + .conditional(HttpConfig.METHOD, HttpRequestMethod.POST, JiraSourceConfig.BODY) + .conditional(HttpConfig.FORMAT, HttpConfig.ResponseFormat.JSON, SeaTunnelSchema.SCHEMA) .optional(JiraSourceConfig.POLL_INTERVAL_MILLS) .optional(JiraSourceConfig.RETRY) .optional(JiraSourceConfig.RETRY_BACKOFF_MAX_MS) diff --git a/seatunnel-connectors-v2/connector-http/connector-http-jira/src/test/java/org/apache/seatunnel/connectors/seatunnel/jira/JiraFactoryTest.java b/seatunnel-connectors-v2/connector-http/connector-http-jira/src/test/java/org/apache/seatunnel/connectors/seatunnel/jira/JiraFactoryTest.java new file mode 100644 index 00000000000..554c9cdb84d --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-jira/src/test/java/org/apache/seatunnel/connectors/seatunnel/jira/JiraFactoryTest.java @@ -0,0 +1,31 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.jira; + +import org.apache.seatunnel.connectors.seatunnel.jira.source.JiraSourceFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class JiraFactoryTest { + + @Test + void optionRule() { + Assertions.assertNotNull((new JiraSourceFactory()).optionRule()); + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-klaviyo/src/main/java/org/apache/seatunnel/connectors/seatunnel/klaviyo/source/KlaviyoSourceFactory.java b/seatunnel-connectors-v2/connector-http/connector-http-klaviyo/src/main/java/org/apache/seatunnel/connectors/seatunnel/klaviyo/source/KlaviyoSourceFactory.java index 56fae935300..fff5633bd01 100644 --- a/seatunnel-connectors-v2/connector-http/connector-http-klaviyo/src/main/java/org/apache/seatunnel/connectors/seatunnel/klaviyo/source/KlaviyoSourceFactory.java +++ b/seatunnel-connectors-v2/connector-http/connector-http-klaviyo/src/main/java/org/apache/seatunnel/connectors/seatunnel/klaviyo/source/KlaviyoSourceFactory.java @@ -17,7 +17,6 @@ package org.apache.seatunnel.connectors.seatunnel.klaviyo.source; -import org.apache.seatunnel.api.configuration.util.Condition; import org.apache.seatunnel.api.configuration.util.OptionRule; import org.apache.seatunnel.api.table.factory.Factory; import org.apache.seatunnel.api.table.factory.TableSourceFactory; @@ -38,19 +37,19 @@ public String factoryIdentifier() { @Override public OptionRule optionRule() { return OptionRule.builder() - .required(KlaviyoSourceConfig.URL) - .required(KlaviyoSourceConfig.PRIVATE_KEY) - .required(KlaviyoSourceConfig.REVISION) - .optional(KlaviyoSourceConfig.METHOD) - .optional(KlaviyoSourceConfig.HEADERS) - .optional(KlaviyoSourceConfig.PARAMS) - .conditional(Condition.of(HttpConfig.METHOD, HttpRequestMethod.POST), KlaviyoSourceConfig.BODY) - .conditional(Condition.of(HttpConfig.FORMAT, "json"), SeaTunnelSchema.SCHEMA) - .optional(KlaviyoSourceConfig.FORMAT) - .optional(KlaviyoSourceConfig.POLL_INTERVAL_MILLS) - .optional(KlaviyoSourceConfig.RETRY) - .optional(KlaviyoSourceConfig.RETRY_BACKOFF_MAX_MS) - .optional(KlaviyoSourceConfig.RETRY_BACKOFF_MULTIPLIER_MS) - .build(); + .required(KlaviyoSourceConfig.URL) + .required(KlaviyoSourceConfig.PRIVATE_KEY) + .required(KlaviyoSourceConfig.REVISION) + .optional(KlaviyoSourceConfig.METHOD) + .optional(KlaviyoSourceConfig.HEADERS) + .optional(KlaviyoSourceConfig.PARAMS) + .optional(KlaviyoSourceConfig.FORMAT) + .conditional(HttpConfig.METHOD, HttpRequestMethod.POST, KlaviyoSourceConfig.BODY) + .conditional(HttpConfig.FORMAT, HttpConfig.ResponseFormat.JSON, SeaTunnelSchema.SCHEMA) + .optional(KlaviyoSourceConfig.POLL_INTERVAL_MILLS) + .optional(KlaviyoSourceConfig.RETRY) + .optional(KlaviyoSourceConfig.RETRY_BACKOFF_MAX_MS) + .optional(KlaviyoSourceConfig.RETRY_BACKOFF_MULTIPLIER_MS) + .build(); } } diff --git a/seatunnel-connectors-v2/connector-http/connector-http-klaviyo/src/test/java/org/apache/seatunnel/connectors/seatunnel/klaviyo/KlaviyoFactoryTest.java b/seatunnel-connectors-v2/connector-http/connector-http-klaviyo/src/test/java/org/apache/seatunnel/connectors/seatunnel/klaviyo/KlaviyoFactoryTest.java new file mode 100644 index 00000000000..c11dae1cf93 --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-klaviyo/src/test/java/org/apache/seatunnel/connectors/seatunnel/klaviyo/KlaviyoFactoryTest.java @@ -0,0 +1,31 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.klaviyo; + +import org.apache.seatunnel.connectors.seatunnel.klaviyo.source.KlaviyoSourceFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class KlaviyoFactoryTest { + + @Test + void optionRule() { + Assertions.assertNotNull((new KlaviyoSourceFactory()).optionRule()); + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-lemlist/src/main/java/org/apache/seatunnel/connectors/seatunnel/lemlist/source/LemlistSourceFactory.java b/seatunnel-connectors-v2/connector-http/connector-http-lemlist/src/main/java/org/apache/seatunnel/connectors/seatunnel/lemlist/source/LemlistSourceFactory.java index c7d4154238d..eaaad2e70a4 100644 --- a/seatunnel-connectors-v2/connector-http/connector-http-lemlist/src/main/java/org/apache/seatunnel/connectors/seatunnel/lemlist/source/LemlistSourceFactory.java +++ b/seatunnel-connectors-v2/connector-http/connector-http-lemlist/src/main/java/org/apache/seatunnel/connectors/seatunnel/lemlist/source/LemlistSourceFactory.java @@ -17,7 +17,6 @@ package org.apache.seatunnel.connectors.seatunnel.lemlist.source; -import org.apache.seatunnel.api.configuration.util.Condition; import org.apache.seatunnel.api.configuration.util.OptionRule; import org.apache.seatunnel.api.table.factory.Factory; import org.apache.seatunnel.api.table.factory.TableSourceFactory; @@ -38,18 +37,18 @@ public String factoryIdentifier() { @Override public OptionRule optionRule() { return OptionRule.builder() - .required(LemlistSourceConfig.URL) - .required(LemlistSourceConfig.PASSWORD) - .optional(LemlistSourceConfig.METHOD) - .optional(LemlistSourceConfig.HEADERS) - .optional(LemlistSourceConfig.PARAMS) - .conditional(Condition.of(HttpConfig.METHOD, HttpRequestMethod.POST), LemlistSourceConfig.BODY) - .conditional(Condition.of(HttpConfig.FORMAT, "json"), SeaTunnelSchema.SCHEMA) - .optional(LemlistSourceConfig.FORMAT) - .optional(LemlistSourceConfig.POLL_INTERVAL_MILLS) - .optional(LemlistSourceConfig.RETRY) - .optional(LemlistSourceConfig.RETRY_BACKOFF_MAX_MS) - .optional(LemlistSourceConfig.RETRY_BACKOFF_MULTIPLIER_MS) - .build(); + .required(LemlistSourceConfig.URL) + .required(LemlistSourceConfig.PASSWORD) + .optional(LemlistSourceConfig.METHOD) + .optional(LemlistSourceConfig.HEADERS) + .optional(LemlistSourceConfig.PARAMS) + .optional(LemlistSourceConfig.FORMAT) + .conditional(HttpConfig.METHOD, HttpRequestMethod.POST, LemlistSourceConfig.BODY) + .conditional(HttpConfig.FORMAT, HttpConfig.ResponseFormat.JSON, SeaTunnelSchema.SCHEMA) + .optional(LemlistSourceConfig.POLL_INTERVAL_MILLS) + .optional(LemlistSourceConfig.RETRY) + .optional(LemlistSourceConfig.RETRY_BACKOFF_MAX_MS) + .optional(LemlistSourceConfig.RETRY_BACKOFF_MULTIPLIER_MS) + .build(); } } diff --git a/seatunnel-connectors-v2/connector-http/connector-http-lemlist/src/test/java/org/apache/seatunnel/connectors/seatunnel/lemlist/LemlistFactoryTest.java b/seatunnel-connectors-v2/connector-http/connector-http-lemlist/src/test/java/org/apache/seatunnel/connectors/seatunnel/lemlist/LemlistFactoryTest.java new file mode 100644 index 00000000000..6ce1fa6e57e --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-lemlist/src/test/java/org/apache/seatunnel/connectors/seatunnel/lemlist/LemlistFactoryTest.java @@ -0,0 +1,31 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.lemlist; + +import org.apache.seatunnel.connectors.seatunnel.lemlist.source.LemlistSourceFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class LemlistFactoryTest { + + @Test + void optionRule() { + Assertions.assertNotNull((new LemlistSourceFactory()).optionRule()); + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-myhours/src/main/java/org/apache/seatunnel/connectors/seatunnel/myhours/source/MyHoursSourceFactory.java b/seatunnel-connectors-v2/connector-http/connector-http-myhours/src/main/java/org/apache/seatunnel/connectors/seatunnel/myhours/source/MyHoursSourceFactory.java index 514f7bb9b3a..f476e0f04f3 100644 --- a/seatunnel-connectors-v2/connector-http/connector-http-myhours/src/main/java/org/apache/seatunnel/connectors/seatunnel/myhours/source/MyHoursSourceFactory.java +++ b/seatunnel-connectors-v2/connector-http/connector-http-myhours/src/main/java/org/apache/seatunnel/connectors/seatunnel/myhours/source/MyHoursSourceFactory.java @@ -17,7 +17,6 @@ package org.apache.seatunnel.connectors.seatunnel.myhours.source; -import org.apache.seatunnel.api.configuration.util.Condition; import org.apache.seatunnel.api.configuration.util.OptionRule; import org.apache.seatunnel.api.table.factory.Factory; import org.apache.seatunnel.api.table.factory.TableSourceFactory; @@ -38,19 +37,19 @@ public String factoryIdentifier() { @Override public OptionRule optionRule() { return OptionRule.builder() - .required(MyHoursSourceConfig.URL) - .required(MyHoursSourceConfig.EMAIL) - .required(MyHoursSourceConfig.PASSWORD) - .optional(MyHoursSourceConfig.METHOD) - .optional(MyHoursSourceConfig.HEADERS) - .optional(MyHoursSourceConfig.PARAMS) - .conditional(Condition.of(HttpConfig.METHOD, HttpRequestMethod.POST), MyHoursSourceConfig.BODY) - .conditional(Condition.of(HttpConfig.FORMAT, "json"), SeaTunnelSchema.SCHEMA) - .optional(MyHoursSourceConfig.FORMAT) - .optional(MyHoursSourceConfig.POLL_INTERVAL_MILLS) - .optional(MyHoursSourceConfig.RETRY) - .optional(MyHoursSourceConfig.RETRY_BACKOFF_MAX_MS) - .optional(MyHoursSourceConfig.RETRY_BACKOFF_MULTIPLIER_MS) - .build(); + .required(MyHoursSourceConfig.URL) + .required(MyHoursSourceConfig.EMAIL) + .required(MyHoursSourceConfig.PASSWORD) + .optional(MyHoursSourceConfig.METHOD) + .optional(MyHoursSourceConfig.HEADERS) + .optional(MyHoursSourceConfig.PARAMS) + .optional(MyHoursSourceConfig.FORMAT) + .conditional(HttpConfig.METHOD, HttpRequestMethod.POST, MyHoursSourceConfig.BODY) + .conditional(HttpConfig.FORMAT, HttpConfig.ResponseFormat.JSON, SeaTunnelSchema.SCHEMA) + .optional(MyHoursSourceConfig.POLL_INTERVAL_MILLS) + .optional(MyHoursSourceConfig.RETRY) + .optional(MyHoursSourceConfig.RETRY_BACKOFF_MAX_MS) + .optional(MyHoursSourceConfig.RETRY_BACKOFF_MULTIPLIER_MS) + .build(); } } diff --git a/seatunnel-connectors-v2/connector-http/connector-http-myhours/src/test/java/org/apache/seatunnel/connectors/seatunnel/myhours/MyHoursFactoryTest.java b/seatunnel-connectors-v2/connector-http/connector-http-myhours/src/test/java/org/apache/seatunnel/connectors/seatunnel/myhours/MyHoursFactoryTest.java new file mode 100644 index 00000000000..567a01d3310 --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-myhours/src/test/java/org/apache/seatunnel/connectors/seatunnel/myhours/MyHoursFactoryTest.java @@ -0,0 +1,31 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.myhours; + +import org.apache.seatunnel.connectors.seatunnel.myhours.source.MyHoursSourceFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class MyHoursFactoryTest { + + @Test + void optionRule() { + Assertions.assertNotNull((new MyHoursSourceFactory()).optionRule()); + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-onesignal/src/main/java/org/apache/seatunnel/connectors/seatunnel/onesignal/source/OneSignalSourceFactory.java b/seatunnel-connectors-v2/connector-http/connector-http-onesignal/src/main/java/org/apache/seatunnel/connectors/seatunnel/onesignal/source/OneSignalSourceFactory.java index ff2b40b23f7..c6c83afe6f2 100644 --- a/seatunnel-connectors-v2/connector-http/connector-http-onesignal/src/main/java/org/apache/seatunnel/connectors/seatunnel/onesignal/source/OneSignalSourceFactory.java +++ b/seatunnel-connectors-v2/connector-http/connector-http-onesignal/src/main/java/org/apache/seatunnel/connectors/seatunnel/onesignal/source/OneSignalSourceFactory.java @@ -17,7 +17,6 @@ package org.apache.seatunnel.connectors.seatunnel.onesignal.source; -import org.apache.seatunnel.api.configuration.util.Condition; import org.apache.seatunnel.api.configuration.util.OptionRule; import org.apache.seatunnel.api.table.factory.Factory; import org.apache.seatunnel.api.table.factory.TableSourceFactory; @@ -38,18 +37,18 @@ public String factoryIdentifier() { @Override public OptionRule optionRule() { return OptionRule.builder() - .required(OneSignalSourceConfig.URL) - .required(OneSignalSourceConfig.PASSWORD) - .optional(OneSignalSourceConfig.METHOD) - .optional(OneSignalSourceConfig.HEADERS) - .optional(OneSignalSourceConfig.PARAMS) - .conditional(Condition.of(HttpConfig.METHOD, HttpRequestMethod.POST), OneSignalSourceConfig.BODY) - .conditional(Condition.of(HttpConfig.FORMAT, "json"), SeaTunnelSchema.SCHEMA) - .optional(OneSignalSourceConfig.FORMAT) - .optional(OneSignalSourceConfig.POLL_INTERVAL_MILLS) - .optional(OneSignalSourceConfig.RETRY) - .optional(OneSignalSourceConfig.RETRY_BACKOFF_MAX_MS) - .optional(OneSignalSourceConfig.RETRY_BACKOFF_MULTIPLIER_MS) - .build(); + .required(OneSignalSourceConfig.URL) + .required(OneSignalSourceConfig.PASSWORD) + .optional(OneSignalSourceConfig.METHOD) + .optional(OneSignalSourceConfig.HEADERS) + .optional(OneSignalSourceConfig.PARAMS) + .optional(OneSignalSourceConfig.FORMAT) + .conditional(HttpConfig.METHOD, HttpRequestMethod.POST, OneSignalSourceConfig.BODY) + .conditional(HttpConfig.FORMAT, HttpConfig.ResponseFormat.JSON, SeaTunnelSchema.SCHEMA) + .optional(OneSignalSourceConfig.POLL_INTERVAL_MILLS) + .optional(OneSignalSourceConfig.RETRY) + .optional(OneSignalSourceConfig.RETRY_BACKOFF_MAX_MS) + .optional(OneSignalSourceConfig.RETRY_BACKOFF_MULTIPLIER_MS) + .build(); } } diff --git a/seatunnel-connectors-v2/connector-http/connector-http-onesignal/src/test/java/org/apache/seatunnel/connectors/seatunnel/onesignal/OneSignalFactoryTest.java b/seatunnel-connectors-v2/connector-http/connector-http-onesignal/src/test/java/org/apache/seatunnel/connectors/seatunnel/onesignal/OneSignalFactoryTest.java new file mode 100644 index 00000000000..d22b16c3f6d --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-onesignal/src/test/java/org/apache/seatunnel/connectors/seatunnel/onesignal/OneSignalFactoryTest.java @@ -0,0 +1,31 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.onesignal; + +import org.apache.seatunnel.connectors.seatunnel.onesignal.source.OneSignalSourceFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class OneSignalFactoryTest { + + @Test + void optionRule() { + Assertions.assertNotNull((new OneSignalSourceFactory()).optionRule()); + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-wechat/src/test/java/org/apache/seatunnel/connectors/seatunnel/wechat/WeChatFactoryTest.java b/seatunnel-connectors-v2/connector-http/connector-http-wechat/src/test/java/org/apache/seatunnel/connectors/seatunnel/wechat/WeChatFactoryTest.java new file mode 100644 index 00000000000..130d6d82e29 --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-wechat/src/test/java/org/apache/seatunnel/connectors/seatunnel/wechat/WeChatFactoryTest.java @@ -0,0 +1,31 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.wechat; + +import org.apache.seatunnel.connectors.seatunnel.wechat.sink.WeChatSinkFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class WeChatFactoryTest { + + @Test + void optionRule() { + Assertions.assertNotNull((new WeChatSinkFactory()).optionRule()); + } +} diff --git a/seatunnel-connectors-v2/connector-iceberg/src/test/java/org/apache/seatunnel/connectors/seatunnel/iceberg/IcebergFactoryTest.java b/seatunnel-connectors-v2/connector-iceberg/src/test/java/org/apache/seatunnel/connectors/seatunnel/iceberg/IcebergFactoryTest.java new file mode 100644 index 00000000000..a4c0753ab8d --- /dev/null +++ b/seatunnel-connectors-v2/connector-iceberg/src/test/java/org/apache/seatunnel/connectors/seatunnel/iceberg/IcebergFactoryTest.java @@ -0,0 +1,31 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.iceberg; + +import org.apache.seatunnel.connectors.seatunnel.iceberg.source.IcebergSourceFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class IcebergFactoryTest { + + @Test + void optionRule() { + Assertions.assertNotNull((new IcebergSourceFactory()).optionRule()); + } +} diff --git a/seatunnel-connectors-v2/connector-influxdb/src/test/java/org/apache/seatunnel/connectors/seatunnel/influxdb/InfluxDBFactoryTest.java b/seatunnel-connectors-v2/connector-influxdb/src/test/java/org/apache/seatunnel/connectors/seatunnel/influxdb/InfluxDBFactoryTest.java new file mode 100644 index 00000000000..2d0404b8a13 --- /dev/null +++ b/seatunnel-connectors-v2/connector-influxdb/src/test/java/org/apache/seatunnel/connectors/seatunnel/influxdb/InfluxDBFactoryTest.java @@ -0,0 +1,33 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.influxdb; + +import org.apache.seatunnel.connectors.seatunnel.influxdb.sink.InfluxDBSinkFactory; +import org.apache.seatunnel.connectors.seatunnel.influxdb.source.InfluxDBSourceFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class InfluxDBFactoryTest { + + @Test + void optionRule() { + Assertions.assertNotNull((new InfluxDBSourceFactory()).optionRule()); + Assertions.assertNotNull((new InfluxDBSinkFactory()).optionRule()); + } +} diff --git a/seatunnel-connectors-v2/connector-iotdb/src/test/java/org/apache/seatunnel/connectors/seatunnel/iotdb/IoTDBFactoryTest.java b/seatunnel-connectors-v2/connector-iotdb/src/test/java/org/apache/seatunnel/connectors/seatunnel/iotdb/IoTDBFactoryTest.java new file mode 100644 index 00000000000..6e5ffdf3fd9 --- /dev/null +++ b/seatunnel-connectors-v2/connector-iotdb/src/test/java/org/apache/seatunnel/connectors/seatunnel/iotdb/IoTDBFactoryTest.java @@ -0,0 +1,33 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.iotdb; + +import org.apache.seatunnel.connectors.seatunnel.iotdb.sink.IoTDBSinkFactory; +import org.apache.seatunnel.connectors.seatunnel.iotdb.source.IoTDBSourceFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class IoTDBFactoryTest { + + @Test + void optionRule() { + Assertions.assertNotNull((new IoTDBSourceFactory()).optionRule()); + Assertions.assertNotNull((new IoTDBSinkFactory()).optionRule()); + } +} diff --git a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcFactoryTest.java b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcFactoryTest.java new file mode 100644 index 00000000000..140e228be80 --- /dev/null +++ b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcFactoryTest.java @@ -0,0 +1,33 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.jdbc; + +import org.apache.seatunnel.connectors.seatunnel.jdbc.sink.JdbcSinkFactory; +import org.apache.seatunnel.connectors.seatunnel.jdbc.source.JdbcSourceFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class JdbcFactoryTest { + + @Test + void optionRule() { + Assertions.assertNotNull((new JdbcSourceFactory()).optionRule()); + Assertions.assertNotNull((new JdbcSinkFactory()).optionRule()); + } +} diff --git a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceFactory.java b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceFactory.java index b115059b853..8a4dce93fe6 100644 --- a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceFactory.java +++ b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceFactory.java @@ -17,7 +17,6 @@ package org.apache.seatunnel.connectors.seatunnel.kafka.source; -import org.apache.seatunnel.api.configuration.util.Condition; import org.apache.seatunnel.api.configuration.util.OptionRule; import org.apache.seatunnel.api.table.factory.Factory; import org.apache.seatunnel.api.table.factory.TableSourceFactory; @@ -38,10 +37,11 @@ public String factoryIdentifier() { public OptionRule optionRule() { return OptionRule.builder() .required(Config.TOPIC, Config.BOOTSTRAP_SERVERS) - .optional(Config.START_MODE, Config.PATTERN, Config.CONSUMER_GROUP, Config.COMMIT_ON_CHECKPOINT, Config.KAFKA_CONFIG_PREFIX, Config.SCHEMA, + .optional(Config.START_MODE, Config.PATTERN, Config.CONSUMER_GROUP, Config.COMMIT_ON_CHECKPOINT, + Config.KAFKA_CONFIG_PREFIX, Config.SCHEMA, Config.FORMAT, Config.KEY_PARTITION_DISCOVERY_INTERVAL_MILLIS) - .conditional(Condition.of(Config.START_MODE, StartMode.TIMESTAMP), Config.START_MODE_TIMESTAMP) - .conditional(Condition.of(Config.START_MODE, StartMode.SPECIFIC_OFFSETS), Config.START_MODE_OFFSETS) + .conditional(Config.START_MODE, StartMode.TIMESTAMP, Config.START_MODE_TIMESTAMP) + .conditional(Config.START_MODE, StartMode.SPECIFIC_OFFSETS, Config.START_MODE_OFFSETS) .build(); } } diff --git a/seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/KafkaFactoryTest.java b/seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/KafkaFactoryTest.java new file mode 100644 index 00000000000..7ab57b02f08 --- /dev/null +++ b/seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/KafkaFactoryTest.java @@ -0,0 +1,33 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.kafka; + +import org.apache.seatunnel.connectors.seatunnel.kafka.sink.KafkaSinkFactory; +import org.apache.seatunnel.connectors.seatunnel.kafka.source.KafkaSourceFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class KafkaFactoryTest { + + @Test + void optionRule() { + Assertions.assertNotNull((new KafkaSourceFactory()).optionRule()); + Assertions.assertNotNull((new KafkaSinkFactory()).optionRule()); + } +} diff --git a/seatunnel-connectors-v2/connector-kudu/src/test/java/org/apache/seatunnel/connectors/seatunnel/kudu/KuduFactoryTest.java b/seatunnel-connectors-v2/connector-kudu/src/test/java/org/apache/seatunnel/connectors/seatunnel/kudu/KuduFactoryTest.java new file mode 100644 index 00000000000..b779cf492ed --- /dev/null +++ b/seatunnel-connectors-v2/connector-kudu/src/test/java/org/apache/seatunnel/connectors/seatunnel/kudu/KuduFactoryTest.java @@ -0,0 +1,33 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.kudu; + +import org.apache.seatunnel.connectors.seatunnel.kudu.sink.KuduSinkFactory; +import org.apache.seatunnel.connectors.seatunnel.kudu.source.KuduSourceFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class KuduFactoryTest { + + @Test + void optionRule() { + Assertions.assertNotNull((new KuduSourceFactory()).optionRule()); + Assertions.assertNotNull((new KuduSinkFactory()).optionRule()); + } +} diff --git a/seatunnel-connectors-v2/connector-mongodb/src/test/java/org/apache/seatunnel/connectors/seatunnel/mongodb/MongodbFactoryTest.java b/seatunnel-connectors-v2/connector-mongodb/src/test/java/org/apache/seatunnel/connectors/seatunnel/mongodb/MongodbFactoryTest.java new file mode 100644 index 00000000000..88053f7d1ab --- /dev/null +++ b/seatunnel-connectors-v2/connector-mongodb/src/test/java/org/apache/seatunnel/connectors/seatunnel/mongodb/MongodbFactoryTest.java @@ -0,0 +1,33 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.mongodb; + +import org.apache.seatunnel.connectors.seatunnel.mongodb.sink.MongodbSinkFactory; +import org.apache.seatunnel.connectors.seatunnel.mongodb.source.MongodbSourceFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class MongodbFactoryTest { + + @Test + void optionRule() { + Assertions.assertNotNull((new MongodbSourceFactory()).optionRule()); + Assertions.assertNotNull((new MongodbSinkFactory()).optionRule()); + } +} diff --git a/seatunnel-connectors-v2/connector-neo4j/src/test/java/org/apache/seatunnel/connectors/seatunnel/neo4j/Neo4jFactoryTest.java b/seatunnel-connectors-v2/connector-neo4j/src/test/java/org/apache/seatunnel/connectors/seatunnel/neo4j/Neo4jFactoryTest.java new file mode 100644 index 00000000000..a77813d7029 --- /dev/null +++ b/seatunnel-connectors-v2/connector-neo4j/src/test/java/org/apache/seatunnel/connectors/seatunnel/neo4j/Neo4jFactoryTest.java @@ -0,0 +1,33 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.neo4j; + +import org.apache.seatunnel.connectors.seatunnel.neo4j.sink.Neo4jSinkFactory; +import org.apache.seatunnel.connectors.seatunnel.neo4j.source.Neo4jSourceFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class Neo4jFactoryTest { + + @Test + void optionRule() { + Assertions.assertNotNull((new Neo4jSourceFactory()).optionRule()); + Assertions.assertNotNull((new Neo4jSinkFactory()).optionRule()); + } +} diff --git a/seatunnel-connectors-v2/connector-openmldb/src/main/java/org/apache/seatunnel/connectors/seatunnel/openmldb/source/OpenMldbSourceFactory.java b/seatunnel-connectors-v2/connector-openmldb/src/main/java/org/apache/seatunnel/connectors/seatunnel/openmldb/source/OpenMldbSourceFactory.java index 749024ef11f..06a83fad938 100644 --- a/seatunnel-connectors-v2/connector-openmldb/src/main/java/org/apache/seatunnel/connectors/seatunnel/openmldb/source/OpenMldbSourceFactory.java +++ b/seatunnel-connectors-v2/connector-openmldb/src/main/java/org/apache/seatunnel/connectors/seatunnel/openmldb/source/OpenMldbSourceFactory.java @@ -17,7 +17,6 @@ package org.apache.seatunnel.connectors.seatunnel.openmldb.source; -import org.apache.seatunnel.api.configuration.util.Condition; import org.apache.seatunnel.api.configuration.util.OptionRule; import org.apache.seatunnel.api.table.factory.Factory; import org.apache.seatunnel.api.table.factory.TableSourceFactory; @@ -35,15 +34,15 @@ public String factoryIdentifier() { @Override public OptionRule optionRule() { return OptionRule.builder() - .required(OpenMldbConfig.CLUSTER_MODE) - .required(OpenMldbConfig.SQL) - .required(OpenMldbConfig.DATABASE) - .optional(OpenMldbConfig.SESSION_TIMEOUT) - .optional(OpenMldbConfig.REQUEST_TIMEOUT) - .conditional(Condition.of(OpenMldbConfig.CLUSTER_MODE, false), - OpenMldbConfig.HOST, OpenMldbConfig.PORT) - .conditional(Condition.of(OpenMldbConfig.CLUSTER_MODE, true), - OpenMldbConfig.ZK_HOST, OpenMldbConfig.ZK_PATH) - .build(); + .required(OpenMldbConfig.CLUSTER_MODE) + .required(OpenMldbConfig.SQL) + .required(OpenMldbConfig.DATABASE) + .optional(OpenMldbConfig.SESSION_TIMEOUT) + .optional(OpenMldbConfig.REQUEST_TIMEOUT) + .conditional(OpenMldbConfig.CLUSTER_MODE, false, + OpenMldbConfig.HOST, OpenMldbConfig.PORT) + .conditional(OpenMldbConfig.CLUSTER_MODE, true, + OpenMldbConfig.ZK_HOST, OpenMldbConfig.ZK_PATH) + .build(); } } diff --git a/seatunnel-connectors-v2/connector-openmldb/src/test/java/org/apache/seatunnel/connectors/seatunnel/openmldb/OpenMldbFactoryTest.java b/seatunnel-connectors-v2/connector-openmldb/src/test/java/org/apache/seatunnel/connectors/seatunnel/openmldb/OpenMldbFactoryTest.java new file mode 100644 index 00000000000..89982ce78db --- /dev/null +++ b/seatunnel-connectors-v2/connector-openmldb/src/test/java/org/apache/seatunnel/connectors/seatunnel/openmldb/OpenMldbFactoryTest.java @@ -0,0 +1,31 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.openmldb; + +import org.apache.seatunnel.connectors.seatunnel.openmldb.source.OpenMldbSourceFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class OpenMldbFactoryTest { + + @Test + void optionRule() { + Assertions.assertNotNull((new OpenMldbSourceFactory()).optionRule()); + } +} diff --git a/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/PulsarSourceFactory.java b/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/PulsarSourceFactory.java index 35184c0dd21..538be437302 100644 --- a/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/PulsarSourceFactory.java +++ b/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/PulsarSourceFactory.java @@ -54,13 +54,13 @@ public String factoryIdentifier() { public OptionRule optionRule() { return OptionRule.builder() .required(SUBSCRIPTION_NAME, CLIENT_SERVICE_URL, ADMIN_SERVICE_URL) + .optional(CURSOR_STARTUP_MODE, CURSOR_STOP_MODE, TOPIC_DISCOVERY_INTERVAL, + POLL_TIMEOUT, POLL_INTERVAL, + POLL_BATCH_SIZE, SeaTunnelSchema.SCHEMA) .exclusive(TOPIC, TOPIC_PATTERN) .conditional(CURSOR_STARTUP_MODE, SourceProperties.StartMode.TIMESTAMP, CURSOR_STARTUP_TIMESTAMP) .conditional(CURSOR_STARTUP_MODE, SourceProperties.StartMode.SUBSCRIPTION, CURSOR_RESET_MODE) .conditional(CURSOR_STOP_MODE, SourceProperties.StopMode.TIMESTAMP, CURSOR_STOP_TIMESTAMP) - .optional(CURSOR_STARTUP_MODE, CURSOR_STOP_MODE, TOPIC_DISCOVERY_INTERVAL, - POLL_TIMEOUT, POLL_INTERVAL, - POLL_BATCH_SIZE, SeaTunnelSchema.SCHEMA) .bundled(AUTH_PLUGIN_CLASS, AUTH_PARAMS) .build(); } diff --git a/seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/source/RabbitmqSourceFactory.java b/seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/source/RabbitmqSourceFactory.java index 413cdaafabd..3bcda92e7c0 100644 --- a/seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/source/RabbitmqSourceFactory.java +++ b/seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/source/RabbitmqSourceFactory.java @@ -57,8 +57,6 @@ public OptionRule optionRule() { HOST, PORT, VIRTUAL_HOST, - USERNAME, - PASSWORD, QUEUE_NAME, SCHEMA ) diff --git a/seatunnel-connectors-v2/connector-rabbitmq/src/test/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/RabbitmqFactoryTest.java b/seatunnel-connectors-v2/connector-rabbitmq/src/test/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/RabbitmqFactoryTest.java new file mode 100644 index 00000000000..36641e585f9 --- /dev/null +++ b/seatunnel-connectors-v2/connector-rabbitmq/src/test/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/RabbitmqFactoryTest.java @@ -0,0 +1,33 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.rabbitmq; + +import org.apache.seatunnel.connectors.seatunnel.rabbitmq.sink.RabbitmqSinkFactory; +import org.apache.seatunnel.connectors.seatunnel.rabbitmq.source.RabbitmqSourceFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class RabbitmqFactoryTest { + + @Test + void optionRule() { + Assertions.assertNotNull((new RabbitmqSourceFactory()).optionRule()); + Assertions.assertNotNull((new RabbitmqSinkFactory()).optionRule()); + } +} diff --git a/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/source/RedisSourceFactory.java b/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/source/RedisSourceFactory.java index 23b506da463..a948ca45fe2 100644 --- a/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/source/RedisSourceFactory.java +++ b/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/source/RedisSourceFactory.java @@ -36,7 +36,8 @@ public String factoryIdentifier() { public OptionRule optionRule() { return OptionRule.builder() .required(RedisConfig.HOST, RedisConfig.PORT, RedisConfig.KEY, RedisConfig.DATA_TYPE) - .optional(RedisConfig.MODE, RedisConfig.HASH_KEY_PARSE_MODE, RedisConfig.AUTH, RedisConfig.USER, RedisConfig.KEY_PATTERN) + .optional(RedisConfig.MODE, RedisConfig.HASH_KEY_PARSE_MODE, RedisConfig.AUTH, RedisConfig.USER, + RedisConfig.KEY_PATTERN) .conditional(RedisConfig.MODE, RedisConfig.RedisMode.CLUSTER, RedisConfig.NODES) .bundled(RedisConfig.FORMAT, SeaTunnelSchema.SCHEMA) .build(); diff --git a/seatunnel-connectors-v2/connector-redis/src/test/java/org/apache/seatunnel/connectors/seatunnel/redis/RedisFactoryTest.java b/seatunnel-connectors-v2/connector-redis/src/test/java/org/apache/seatunnel/connectors/seatunnel/redis/RedisFactoryTest.java new file mode 100644 index 00000000000..ef2b3f0367a --- /dev/null +++ b/seatunnel-connectors-v2/connector-redis/src/test/java/org/apache/seatunnel/connectors/seatunnel/redis/RedisFactoryTest.java @@ -0,0 +1,33 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.redis; + +import org.apache.seatunnel.connectors.seatunnel.redis.sink.RedisSinkFactory; +import org.apache.seatunnel.connectors.seatunnel.redis.source.RedisSourceFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class RedisFactoryTest { + + @Test + void optionRule() { + Assertions.assertNotNull((new RedisSourceFactory()).optionRule()); + Assertions.assertNotNull((new RedisSinkFactory()).optionRule()); + } +} diff --git a/seatunnel-connectors-v2/connector-sentry/src/test/java/org/apache/seatunnel/connectors/seatunnel/sentry/SentryFactoryTest.java b/seatunnel-connectors-v2/connector-sentry/src/test/java/org/apache/seatunnel/connectors/seatunnel/sentry/SentryFactoryTest.java new file mode 100644 index 00000000000..8f7f0cb12d7 --- /dev/null +++ b/seatunnel-connectors-v2/connector-sentry/src/test/java/org/apache/seatunnel/connectors/seatunnel/sentry/SentryFactoryTest.java @@ -0,0 +1,31 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.sentry; + +import org.apache.seatunnel.connectors.seatunnel.sentry.sink.SentrySinkFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class SentryFactoryTest { + + @Test + void optionRule() { + Assertions.assertNotNull((new SentrySinkFactory()).optionRule()); + } +} diff --git a/seatunnel-connectors-v2/connector-slack/src/test/java/org/apache/seatunnel/connectors/seatunnel/slack/SlackFactoryTest.java b/seatunnel-connectors-v2/connector-slack/src/test/java/org/apache/seatunnel/connectors/seatunnel/slack/SlackFactoryTest.java new file mode 100644 index 00000000000..f7e553535d7 --- /dev/null +++ b/seatunnel-connectors-v2/connector-slack/src/test/java/org/apache/seatunnel/connectors/seatunnel/slack/SlackFactoryTest.java @@ -0,0 +1,31 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.slack; + +import org.apache.seatunnel.connectors.seatunnel.slack.sink.SlackSinkFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class SlackFactoryTest { + + @Test + void optionRule() { + Assertions.assertNotNull((new SlackSinkFactory()).optionRule()); + } +} diff --git a/seatunnel-connectors-v2/connector-socket/src/test/java/org/apache/seatunnel/connectors/seatunnel/socket/SocketFactoryTest.java b/seatunnel-connectors-v2/connector-socket/src/test/java/org/apache/seatunnel/connectors/seatunnel/socket/SocketFactoryTest.java new file mode 100644 index 00000000000..6ddc77bfe3d --- /dev/null +++ b/seatunnel-connectors-v2/connector-socket/src/test/java/org/apache/seatunnel/connectors/seatunnel/socket/SocketFactoryTest.java @@ -0,0 +1,33 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.socket; + +import org.apache.seatunnel.connectors.seatunnel.socket.sink.SocketSinkFactory; +import org.apache.seatunnel.connectors.seatunnel.socket.source.SocketSourceFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class SocketFactoryTest { + + @Test + void optionRule() { + Assertions.assertNotNull((new SocketSourceFactory()).optionRule()); + Assertions.assertNotNull((new SocketSinkFactory()).optionRule()); + } +} diff --git a/seatunnel-connectors-v2/connector-starrocks/src/test/java/org/apache/seatunnel/connectors/seatunnel/starrocks/StarRocksFactoryTest.java b/seatunnel-connectors-v2/connector-starrocks/src/test/java/org/apache/seatunnel/connectors/seatunnel/starrocks/StarRocksFactoryTest.java new file mode 100644 index 00000000000..9bc934aac95 --- /dev/null +++ b/seatunnel-connectors-v2/connector-starrocks/src/test/java/org/apache/seatunnel/connectors/seatunnel/starrocks/StarRocksFactoryTest.java @@ -0,0 +1,31 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.starrocks; + +import org.apache.seatunnel.connectors.seatunnel.starrocks.sink.StarRocksSinkFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class StarRocksFactoryTest { + + @Test + void optionRule() { + Assertions.assertNotNull((new StarRocksSinkFactory()).optionRule()); + } +} diff --git a/seatunnel-connectors-v2/connector-tablestore/src/test/java/org/apache/seatunnel/connectors/seatunnel/tablestore/TablestoreFactoryTest.java b/seatunnel-connectors-v2/connector-tablestore/src/test/java/org/apache/seatunnel/connectors/seatunnel/tablestore/TablestoreFactoryTest.java new file mode 100644 index 00000000000..58920f8a51d --- /dev/null +++ b/seatunnel-connectors-v2/connector-tablestore/src/test/java/org/apache/seatunnel/connectors/seatunnel/tablestore/TablestoreFactoryTest.java @@ -0,0 +1,31 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.tablestore; + +import org.apache.seatunnel.connectors.seatunnel.tablestore.sink.TablestoreSinkFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class TablestoreFactoryTest { + + @Test + void optionRule() { + Assertions.assertNotNull((new TablestoreSinkFactory()).optionRule()); + } +} diff --git a/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/config/JobConfig.java b/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/config/JobConfig.java index ced337b9b02..30469300a2d 100644 --- a/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/config/JobConfig.java +++ b/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/config/JobConfig.java @@ -18,6 +18,7 @@ package org.apache.seatunnel.engine.common.config; import org.apache.seatunnel.api.common.JobContext; +import org.apache.seatunnel.api.env.EnvCommonOptions; import org.apache.seatunnel.engine.common.serializeable.ConfigDataSerializerHook; import com.hazelcast.nio.ObjectDataInput; @@ -31,7 +32,7 @@ @Data public class JobConfig implements IdentifiedDataSerializable { - private String name; + private String name = EnvCommonOptions.JOB_NAME.defaultValue(); private JobContext jobContext; private Map envOptions = new HashMap<>(); diff --git a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/AbstractSeaTunnelServerTest.java b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/AbstractSeaTunnelServerTest.java index af945ee23d0..af45774a6f4 100644 --- a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/AbstractSeaTunnelServerTest.java +++ b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/AbstractSeaTunnelServerTest.java @@ -29,7 +29,7 @@ @Slf4j @TestInstance(TestInstance.Lifecycle.PER_CLASS) -public abstract class AbstractSeaTunnelServerTest { +public abstract class AbstractSeaTunnelServerTest { protected SeaTunnelServer server; @@ -41,8 +41,9 @@ public abstract class AbstractSeaTunnelServerTest { @BeforeAll public void before() { + String name = ((T) this).getClass().getName(); instance = SeaTunnelServerStarter.createHazelcastInstance( - TestUtils.getClusterName("AbstractSeaTunnelServerTest_" + System.currentTimeMillis())); + TestUtils.getClusterName("AbstractSeaTunnelServerTest_" + name)); nodeEngine = instance.node.nodeEngine; server = nodeEngine.getService(SeaTunnelServer.SERVICE_NAME); LOGGER = nodeEngine.getLogger(AbstractSeaTunnelServerTest.class); diff --git a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/master/JobMasterTest.java b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/master/JobMasterTest.java index 53108b79001..c935f0be194 100644 --- a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/master/JobMasterTest.java +++ b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/master/JobMasterTest.java @@ -117,7 +117,7 @@ public void testHandleCheckpointTimeout() throws Exception { JobMaster jobMaster = server.getCoordinatorService().getJobMaster(JOB_ID); // waiting for job status turn to running - await().atMost(60000, TimeUnit.MILLISECONDS) + await().atMost(120000, TimeUnit.MILLISECONDS) .untilAsserted(() -> Assertions.assertEquals(JobStatus.RUNNING, jobMaster.getJobStatus())); // call checkpoint timeout @@ -127,7 +127,7 @@ public void testHandleCheckpointTimeout() throws Exception { Thread.sleep(5000); // test job still run - await().atMost(60000, TimeUnit.MILLISECONDS) + await().atMost(120000, TimeUnit.MILLISECONDS) .untilAsserted(() -> Assertions.assertEquals(JobStatus.RUNNING, jobMaster.getJobStatus())); PassiveCompletableFuture jobMasterCompleteFuture = jobMaster.getJobMasterCompleteFuture(); @@ -135,7 +135,7 @@ public void testHandleCheckpointTimeout() throws Exception { jobMaster.cancelJob(); // test job turn to complete - await().atMost(60000, TimeUnit.MILLISECONDS) + await().atMost(120000, TimeUnit.MILLISECONDS) .untilAsserted(() -> Assertions.assertTrue( jobMasterCompleteFuture.isDone() && JobStatus.CANCELED.equals(jobMasterCompleteFuture.get()))); diff --git a/seatunnel-plugin-discovery/src/main/java/org/apache/seatunnel/plugin/discovery/AbstractPluginDiscovery.java b/seatunnel-plugin-discovery/src/main/java/org/apache/seatunnel/plugin/discovery/AbstractPluginDiscovery.java index 412825b2bb0..9d5f41bfc2f 100644 --- a/seatunnel-plugin-discovery/src/main/java/org/apache/seatunnel/plugin/discovery/AbstractPluginDiscovery.java +++ b/seatunnel-plugin-discovery/src/main/java/org/apache/seatunnel/plugin/discovery/AbstractPluginDiscovery.java @@ -50,7 +50,6 @@ import java.net.URL; import java.net.URLClassLoader; import java.nio.file.Path; -import java.util.ArrayList; import java.util.Collection; import java.util.Collections; import java.util.HashMap; @@ -203,39 +202,21 @@ public T createPluginInstance(PluginIdentifier pluginIdentifier, Collection /** * Get all support plugin already in SEATUNNEL_HOME, only support connector-v2 - * - * @param pluginType choose which type plugin should be returned * @return the all plugin identifier of the engine */ - @SuppressWarnings("unchecked") - public @Nonnull List getAllPlugin(PluginType pluginType) throws IOException { - List files = FileUtils.searchJarFiles(pluginDir); - List plugins = new ArrayList<>(); - List factories; - if (pluginType.equals(PluginType.SOURCE)) { - factories = FactoryUtil.discoverFactories(new URLClassLoader(files.toArray(new URL[0])), TableSourceFactory.class); - } else if (pluginType.equals(PluginType.SINK)) { - factories = FactoryUtil.discoverFactories(new URLClassLoader(files.toArray(new URL[0])), TableSinkFactory.class); - } else if (pluginType.equals(PluginType.TRANSFORM)) { - factories = FactoryUtil.discoverFactories(new URLClassLoader(files.toArray(new URL[0])), TableTransformFactory.class); + @SuppressWarnings("checkstyle:WhitespaceAfter") + public Map> getAllPlugin() throws IOException { + List factories; + if (this.pluginDir.toFile().exists()) { + log.info("load plugin from plugin dir: {}", this.pluginDir); + List files = FileUtils.searchJarFiles(this.pluginDir); + factories = FactoryUtil.discoverFactories(new URLClassLoader((URL[]) files.toArray(new URL[0]))); } else { - throw new IllegalArgumentException("Unsupported plugin type: " + pluginType); + log.info("plugin dir: {} not exists, load plugin from classpath", this.pluginDir); + factories = FactoryUtil.discoverFactories(Thread.currentThread().getContextClassLoader()); } - factories.forEach(plugin -> { - plugins.add(PluginIdentifier.of("seatunnel", pluginType.getType(), ((Factory) plugin).factoryIdentifier())); - }); - return plugins; - } - /** - * Get all support plugin already in SEATUNNEL_HOME, only support connector-v2 - * @return the all plugin identifier of the engine - */ - @SuppressWarnings("checkstyle:WhitespaceAfter") - public Map> getAllPlugin() throws IOException { - List files = FileUtils.searchJarFiles(this.pluginDir); Map> plugins = new HashMap<>(); - List factories = FactoryUtil.discoverFactories(new URLClassLoader((URL[]) files.toArray(new URL[0]))); factories.forEach(plugin -> { if (TableSourceFactory.class.isAssignableFrom(plugin.getClass())) { diff --git a/seatunnel-transforms-v2/pom.xml b/seatunnel-transforms-v2/pom.xml index 9ec96b4321a..328c932224a 100644 --- a/seatunnel-transforms-v2/pom.xml +++ b/seatunnel-transforms-v2/pom.xml @@ -45,6 +45,17 @@ org.apache.seatunnel seatunnel-api + + org.junit.jupiter + junit-jupiter-engine + test + + + + org.junit.jupiter + junit-jupiter-params + test + diff --git a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/ReplaceTransform.java b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/ReplaceTransform.java index a7101558ee3..1db405ab437 100644 --- a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/ReplaceTransform.java +++ b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/ReplaceTransform.java @@ -57,7 +57,7 @@ public class ReplaceTransform extends SingleFieldOutputTransform { public static final Option KEY_REPLACE_FIRST = Options.key("replace_first") .booleanType() - .defaultValue(false) + .noDefaultValue() .withDescription("Replace the first match string"); private int inputFieldIndex; diff --git a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/ReplaceTransformFactory.java b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/ReplaceTransformFactory.java index 88372e6e35b..20030f2d032 100644 --- a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/ReplaceTransformFactory.java +++ b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/ReplaceTransformFactory.java @@ -23,7 +23,6 @@ import static org.apache.seatunnel.transform.ReplaceTransform.KEY_REPLACE_FIELD; import static org.apache.seatunnel.transform.ReplaceTransform.KEY_REPLACE_FIRST; -import org.apache.seatunnel.api.configuration.util.Condition; import org.apache.seatunnel.api.configuration.util.OptionRule; import org.apache.seatunnel.api.table.factory.Factory; import org.apache.seatunnel.api.table.factory.TableTransformFactory; @@ -40,8 +39,9 @@ public String factoryIdentifier() { @Override public OptionRule optionRule() { return OptionRule.builder() - .required(KEY_REPLACE_FIELD, KEY_PATTERN, KEY_REPLACEMENT) - .conditional(Condition.of(KEY_IS_REGEX, true), KEY_REPLACE_FIRST) - .build(); + .required(KEY_REPLACE_FIELD, KEY_PATTERN, KEY_REPLACEMENT) + .optional(KEY_IS_REGEX) + .conditional(KEY_IS_REGEX, true, KEY_REPLACE_FIRST) + .build(); } } diff --git a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/ReplaceTransformFactoryTest.java b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/ReplaceTransformFactoryTest.java new file mode 100644 index 00000000000..5bc9267c425 --- /dev/null +++ b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/ReplaceTransformFactoryTest.java @@ -0,0 +1,30 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.transform; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +public class ReplaceTransformFactoryTest { + + @Test + public void testOptionRule() throws Exception { + ReplaceTransformFactory replaceTransformFactory = new ReplaceTransformFactory(); + Assertions.assertNotNull(replaceTransformFactory.optionRule()); + } +}