This is an automated email from the ASF dual-hosted git repository. wanglijie pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/flink-connector-jdbc.git
commit 396f205bcbed133d1e7dc0d6c3ce9487dcbc1dd7 Author: Joao Boto <[email protected]> AuthorDate: Thu Apr 13 15:00:33 2023 +0200 [FLINK-31649] Fix archunit connectors options must be public --- .../b10c194a-0517-4744-830f-81abfb71ab18 | 2 -- .../connector/jdbc/internal/JdbcOutputFormat.java | 6 ++-- ...ons.java => InternalJdbcConnectionOptions.java} | 14 ++++---- .../jdbc/table/JdbcDynamicTableFactory.java | 13 ++++---- .../connector/jdbc/table/JdbcDynamicTableSink.java | 6 ++-- .../jdbc/table/JdbcDynamicTableSource.java | 6 ++-- .../jdbc/table/JdbcOutputFormatBuilder.java | 6 ++-- .../jdbc/table/JdbcRowDataLookupFunction.java | 4 +-- .../connector/jdbc/internal/JdbcFullTest.java | 10 +++--- .../jdbc/internal/JdbcTableOutputFormatTest.java | 12 +++---- .../jdbc/table/JdbcAppendOnlyWriterTest.java | 4 +-- .../jdbc/table/JdbcDynamicTableFactoryTest.java | 30 ++++++++--------- .../connector/jdbc/table/JdbcOutputFormatTest.java | 38 +++++++++++----------- .../jdbc/table/JdbcRowDataInputFormatTest.java | 4 +-- .../jdbc/table/JdbcRowDataLookupFunctionTest.java | 6 ++-- 15 files changed, 80 insertions(+), 81 deletions(-) diff --git a/flink-connector-jdbc/archunit-violations/b10c194a-0517-4744-830f-81abfb71ab18 b/flink-connector-jdbc/archunit-violations/b10c194a-0517-4744-830f-81abfb71ab18 index 400c942..e69de29 100644 --- a/flink-connector-jdbc/archunit-violations/b10c194a-0517-4744-830f-81abfb71ab18 +++ b/flink-connector-jdbc/archunit-violations/b10c194a-0517-4744-830f-81abfb71ab18 @@ -1,2 +0,0 @@ -Class <org.apache.flink.connector.jdbc.internal.options.JdbcConnectorOptions> does not reside in a package 'org.apache.flink..table' in (JdbcConnectorOptions.java:0) -org.apache.flink.connector.jdbc.internal.options.JdbcConnectorOptions does not satisfy: annotated with @PublicEvolving or annotated with @Public \ No newline at end of file diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/JdbcOutputFormat.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/JdbcOutputFormat.java index a869f6c..1ca8b63 100644 --- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/JdbcOutputFormat.java +++ b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/JdbcOutputFormat.java @@ -33,7 +33,7 @@ import org.apache.flink.connector.jdbc.JdbcStatementBuilder; import org.apache.flink.connector.jdbc.internal.connection.JdbcConnectionProvider; import org.apache.flink.connector.jdbc.internal.connection.SimpleJdbcConnectionProvider; import org.apache.flink.connector.jdbc.internal.executor.JdbcBatchStatementExecutor; -import org.apache.flink.connector.jdbc.internal.options.JdbcConnectorOptions; +import org.apache.flink.connector.jdbc.internal.options.InternalJdbcConnectionOptions; import org.apache.flink.connector.jdbc.internal.options.JdbcDmlOptions; import org.apache.flink.connector.jdbc.statement.FieldNamedPreparedStatementImpl; import org.apache.flink.connector.jdbc.utils.JdbcUtils; @@ -284,7 +284,7 @@ public class JdbcOutputFormat<In, JdbcIn, JdbcExec extends JdbcBatchStatementExe /** Builder for a {@link JdbcOutputFormat}. */ public static class Builder { - private JdbcConnectorOptions options; + private InternalJdbcConnectionOptions options; private String[] fieldNames; private String[] keyFields; private int[] fieldTypes; @@ -292,7 +292,7 @@ public class JdbcOutputFormat<In, JdbcIn, JdbcExec extends JdbcBatchStatementExe JdbcExecutionOptions.builder(); /** required, jdbc options. */ - public Builder setOptions(JdbcConnectorOptions options) { + public Builder setOptions(InternalJdbcConnectionOptions options) { this.options = options; return this; } diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/options/JdbcConnectorOptions.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/options/InternalJdbcConnectionOptions.java similarity index 94% rename from flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/options/JdbcConnectorOptions.java rename to flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/options/InternalJdbcConnectionOptions.java index 0b2562a..6d9aeca 100644 --- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/options/JdbcConnectorOptions.java +++ b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/options/InternalJdbcConnectionOptions.java @@ -32,7 +32,7 @@ import static org.apache.flink.util.Preconditions.checkNotNull; /** Options for the JDBC connector. */ @Internal -public class JdbcConnectorOptions extends JdbcConnectionOptions { +public class InternalJdbcConnectionOptions extends JdbcConnectionOptions { private static final long serialVersionUID = 1L; @@ -40,7 +40,7 @@ public class JdbcConnectorOptions extends JdbcConnectionOptions { private final JdbcDialect dialect; private final @Nullable Integer parallelism; - private JdbcConnectorOptions( + private InternalJdbcConnectionOptions( String dbURL, String tableName, @Nullable String driverName, @@ -73,8 +73,8 @@ public class JdbcConnectorOptions extends JdbcConnectionOptions { @Override public boolean equals(Object o) { - if (o instanceof JdbcConnectorOptions) { - JdbcConnectorOptions options = (JdbcConnectorOptions) o; + if (o instanceof InternalJdbcConnectionOptions) { + InternalJdbcConnectionOptions options = (InternalJdbcConnectionOptions) o; return Objects.equals(url, options.url) && Objects.equals(tableName, options.tableName) && Objects.equals(driverName, options.driverName) @@ -103,7 +103,7 @@ public class JdbcConnectorOptions extends JdbcConnectionOptions { connectionCheckTimeoutSeconds); } - /** Builder of {@link JdbcConnectorOptions}. */ + /** Builder of {@link InternalJdbcConnectionOptions}. */ public static class Builder { private ClassLoader classLoader; private String dbURL; @@ -181,7 +181,7 @@ public class JdbcConnectorOptions extends JdbcConnectionOptions { return this; } - public JdbcConnectorOptions build() { + public InternalJdbcConnectionOptions build() { checkNotNull(dbURL, "No dbURL supplied."); checkNotNull(tableName, "No tableName supplied."); if (this.dialect == null) { @@ -197,7 +197,7 @@ public class JdbcConnectorOptions extends JdbcConnectionOptions { () -> new NullPointerException("No driverName supplied.")); } - return new JdbcConnectorOptions( + return new InternalJdbcConnectionOptions( dialect.appendDefaultUrlProperties(dbURL), tableName, driverName, diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableFactory.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableFactory.java index bff5a49..3434f60 100644 --- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableFactory.java +++ b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableFactory.java @@ -25,7 +25,7 @@ import org.apache.flink.configuration.ReadableConfig; import org.apache.flink.connector.jdbc.JdbcExecutionOptions; import org.apache.flink.connector.jdbc.dialect.JdbcDialect; import org.apache.flink.connector.jdbc.dialect.JdbcDialectLoader; -import org.apache.flink.connector.jdbc.internal.options.JdbcConnectorOptions; +import org.apache.flink.connector.jdbc.internal.options.InternalJdbcConnectionOptions; import org.apache.flink.connector.jdbc.internal.options.JdbcDmlOptions; import org.apache.flink.connector.jdbc.internal.options.JdbcReadOptions; import org.apache.flink.table.connector.sink.DynamicTableSink; @@ -90,7 +90,8 @@ public class JdbcDynamicTableFactory implements DynamicTableSourceFactory, Dynam validateConfigOptions(config, context.getClassLoader()); validateDataTypeWithJdbcDialect( context.getPhysicalRowDataType(), config.get(URL), context.getClassLoader()); - JdbcConnectorOptions jdbcOptions = getJdbcOptions(config, context.getClassLoader()); + InternalJdbcConnectionOptions jdbcOptions = + getJdbcOptions(config, context.getClassLoader()); return new JdbcDynamicTableSink( jdbcOptions, @@ -126,11 +127,11 @@ public class JdbcDynamicTableFactory implements DynamicTableSourceFactory, Dynam dialect.validate((RowType) dataType.getLogicalType()); } - private JdbcConnectorOptions getJdbcOptions( + private InternalJdbcConnectionOptions getJdbcOptions( ReadableConfig readableConfig, ClassLoader classLoader) { final String url = readableConfig.get(URL); - final JdbcConnectorOptions.Builder builder = - JdbcConnectorOptions.builder() + final InternalJdbcConnectionOptions.Builder builder = + InternalJdbcConnectionOptions.builder() .setClassLoader(classLoader) .setDBUrl(url) .setTableName(readableConfig.get(TABLE_NAME)) @@ -169,7 +170,7 @@ public class JdbcDynamicTableFactory implements DynamicTableSourceFactory, Dynam } private JdbcDmlOptions getJdbcDmlOptions( - JdbcConnectorOptions jdbcOptions, DataType dataType, int[] primaryKeyIndexes) { + InternalJdbcConnectionOptions jdbcOptions, DataType dataType, int[] primaryKeyIndexes) { String[] keyFields = Arrays.stream(primaryKeyIndexes) diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableSink.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableSink.java index 03256ff..6140a22 100644 --- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableSink.java +++ b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableSink.java @@ -22,7 +22,7 @@ import org.apache.flink.annotation.Internal; import org.apache.flink.api.common.typeinfo.TypeInformation; import org.apache.flink.connector.jdbc.JdbcExecutionOptions; import org.apache.flink.connector.jdbc.internal.GenericJdbcSinkFunction; -import org.apache.flink.connector.jdbc.internal.options.JdbcConnectorOptions; +import org.apache.flink.connector.jdbc.internal.options.InternalJdbcConnectionOptions; import org.apache.flink.connector.jdbc.internal.options.JdbcDmlOptions; import org.apache.flink.table.connector.ChangelogMode; import org.apache.flink.table.connector.sink.DynamicTableSink; @@ -39,14 +39,14 @@ import static org.apache.flink.util.Preconditions.checkState; @Internal public class JdbcDynamicTableSink implements DynamicTableSink { - private final JdbcConnectorOptions jdbcOptions; + private final InternalJdbcConnectionOptions jdbcOptions; private final JdbcExecutionOptions executionOptions; private final JdbcDmlOptions dmlOptions; private final DataType physicalRowDataType; private final String dialectName; public JdbcDynamicTableSink( - JdbcConnectorOptions jdbcOptions, + InternalJdbcConnectionOptions jdbcOptions, JdbcExecutionOptions executionOptions, JdbcDmlOptions dmlOptions, DataType physicalRowDataType) { diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableSource.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableSource.java index e27cf63..48e1702 100644 --- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableSource.java +++ b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableSource.java @@ -20,7 +20,7 @@ package org.apache.flink.connector.jdbc.table; import org.apache.flink.annotation.Internal; import org.apache.flink.connector.jdbc.dialect.JdbcDialect; -import org.apache.flink.connector.jdbc.internal.options.JdbcConnectorOptions; +import org.apache.flink.connector.jdbc.internal.options.InternalJdbcConnectionOptions; import org.apache.flink.connector.jdbc.internal.options.JdbcReadOptions; import org.apache.flink.connector.jdbc.split.CompositeJdbcParameterValuesProvider; import org.apache.flink.connector.jdbc.split.JdbcGenericParameterValuesProvider; @@ -68,7 +68,7 @@ public class JdbcDynamicTableSource SupportsFilterPushDown { private static final Logger LOG = LoggerFactory.getLogger(JdbcDynamicTableSource.class); - private final JdbcConnectorOptions options; + private final InternalJdbcConnectionOptions options; private final JdbcReadOptions readOptions; private final int lookupMaxRetryTimes; @Nullable private final LookupCache cache; @@ -79,7 +79,7 @@ public class JdbcDynamicTableSource private Serializable[] pushdownParams = new Serializable[0]; public JdbcDynamicTableSource( - JdbcConnectorOptions options, + InternalJdbcConnectionOptions options, JdbcReadOptions readOptions, int lookupMaxRetryTimes, @Nullable LookupCache cache, diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcOutputFormatBuilder.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcOutputFormatBuilder.java index df2031b..8427f29 100644 --- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcOutputFormatBuilder.java +++ b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcOutputFormatBuilder.java @@ -31,7 +31,7 @@ import org.apache.flink.connector.jdbc.internal.executor.TableBufferReducedState import org.apache.flink.connector.jdbc.internal.executor.TableBufferedStatementExecutor; import org.apache.flink.connector.jdbc.internal.executor.TableInsertOrUpdateStatementExecutor; import org.apache.flink.connector.jdbc.internal.executor.TableSimpleStatementExecutor; -import org.apache.flink.connector.jdbc.internal.options.JdbcConnectorOptions; +import org.apache.flink.connector.jdbc.internal.options.InternalJdbcConnectionOptions; import org.apache.flink.connector.jdbc.internal.options.JdbcDmlOptions; import org.apache.flink.connector.jdbc.statement.FieldNamedPreparedStatement; import org.apache.flink.table.data.GenericRowData; @@ -53,7 +53,7 @@ public class JdbcOutputFormatBuilder implements Serializable { private static final long serialVersionUID = 1L; - private JdbcConnectorOptions jdbcOptions; + private InternalJdbcConnectionOptions jdbcOptions; private JdbcExecutionOptions executionOptions; private JdbcDmlOptions dmlOptions; private TypeInformation<RowData> rowDataTypeInformation; @@ -61,7 +61,7 @@ public class JdbcOutputFormatBuilder implements Serializable { public JdbcOutputFormatBuilder() {} - public JdbcOutputFormatBuilder setJdbcOptions(JdbcConnectorOptions jdbcOptions) { + public JdbcOutputFormatBuilder setJdbcOptions(InternalJdbcConnectionOptions jdbcOptions) { this.jdbcOptions = jdbcOptions; return this; } diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcRowDataLookupFunction.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcRowDataLookupFunction.java index cb344b2..4d327b4 100644 --- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcRowDataLookupFunction.java +++ b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcRowDataLookupFunction.java @@ -24,7 +24,7 @@ import org.apache.flink.connector.jdbc.converter.JdbcRowConverter; import org.apache.flink.connector.jdbc.dialect.JdbcDialect; import org.apache.flink.connector.jdbc.internal.connection.JdbcConnectionProvider; import org.apache.flink.connector.jdbc.internal.connection.SimpleJdbcConnectionProvider; -import org.apache.flink.connector.jdbc.internal.options.JdbcConnectorOptions; +import org.apache.flink.connector.jdbc.internal.options.InternalJdbcConnectionOptions; import org.apache.flink.connector.jdbc.statement.FieldNamedPreparedStatement; import org.apache.flink.table.data.RowData; import org.apache.flink.table.functions.FunctionContext; @@ -66,7 +66,7 @@ public class JdbcRowDataLookupFunction extends LookupFunction { private transient FieldNamedPreparedStatement statement; public JdbcRowDataLookupFunction( - JdbcConnectorOptions options, + InternalJdbcConnectionOptions options, int maxRetryTimes, String[] fieldNames, DataType[] fieldTypes, diff --git a/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/internal/JdbcFullTest.java b/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/internal/JdbcFullTest.java index 057bbb0..d41b685 100644 --- a/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/internal/JdbcFullTest.java +++ b/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/internal/JdbcFullTest.java @@ -23,14 +23,13 @@ import org.apache.flink.api.common.functions.RuntimeContext; import org.apache.flink.api.java.DataSet; import org.apache.flink.api.java.ExecutionEnvironment; import org.apache.flink.api.java.tuple.Tuple2; -import org.apache.flink.connector.jdbc.JdbcConnectionOptions; import org.apache.flink.connector.jdbc.JdbcDataTestBase; import org.apache.flink.connector.jdbc.JdbcExecutionOptions; import org.apache.flink.connector.jdbc.JdbcInputFormat; import org.apache.flink.connector.jdbc.JdbcStatementBuilder; import org.apache.flink.connector.jdbc.internal.connection.SimpleJdbcConnectionProvider; import org.apache.flink.connector.jdbc.internal.executor.JdbcBatchStatementExecutor; -import org.apache.flink.connector.jdbc.internal.options.JdbcConnectorOptions; +import org.apache.flink.connector.jdbc.internal.options.InternalJdbcConnectionOptions; import org.apache.flink.connector.jdbc.split.JdbcNumericBetweenParametersProvider; import org.apache.flink.types.Row; @@ -79,7 +78,7 @@ class JdbcFullTest extends JdbcDataTestBase { JdbcOutputFormat jdbcOutputFormat = JdbcOutputFormat.builder() .setOptions( - JdbcConnectorOptions.builder() + InternalJdbcConnectionOptions.builder() .setDBUrl(getMetadata().getJdbcUrl()) .setTableName(OUTPUT_TABLE) .build()) @@ -136,8 +135,9 @@ class JdbcFullTest extends JdbcDataTestBase { // NOTE: in this case (with Derby driver) setSqlTypes could be skipped, but // some databases don't null values correctly when no column type was specified // in PreparedStatement.setObject (see its javadoc for more details) - JdbcConnectionOptions connectionOptions = - new JdbcConnectionOptions.JdbcConnectionOptionsBuilder() + org.apache.flink.connector.jdbc.JdbcConnectionOptions connectionOptions = + new org.apache.flink.connector.jdbc.JdbcConnectionOptions + .JdbcConnectionOptionsBuilder() .withUrl(getMetadata().getJdbcUrl()) .withDriverName(getMetadata().getDriverClass()) .build(); diff --git a/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/internal/JdbcTableOutputFormatTest.java b/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/internal/JdbcTableOutputFormatTest.java index 11eca03..d0caddf 100644 --- a/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/internal/JdbcTableOutputFormatTest.java +++ b/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/internal/JdbcTableOutputFormatTest.java @@ -25,7 +25,7 @@ import org.apache.flink.connector.jdbc.JdbcDataTestBase; import org.apache.flink.connector.jdbc.JdbcExecutionOptions; import org.apache.flink.connector.jdbc.internal.connection.SimpleJdbcConnectionProvider; import org.apache.flink.connector.jdbc.internal.executor.JdbcBatchStatementExecutor; -import org.apache.flink.connector.jdbc.internal.options.JdbcConnectorOptions; +import org.apache.flink.connector.jdbc.internal.options.InternalJdbcConnectionOptions; import org.apache.flink.connector.jdbc.internal.options.JdbcDmlOptions; import org.apache.flink.types.Row; @@ -65,8 +65,8 @@ public class JdbcTableOutputFormatTest extends JdbcDataTestBase { @Test void testUpsertFormatCloseBeforeOpen() throws Exception { - JdbcConnectorOptions options = - JdbcConnectorOptions.builder() + InternalJdbcConnectionOptions options = + InternalJdbcConnectionOptions.builder() .setDBUrl(getMetadata().getJdbcUrl()) .setTableName(OUTPUT_TABLE) .build(); @@ -101,7 +101,7 @@ public class JdbcTableOutputFormatTest extends JdbcDataTestBase { format = new TableJdbcUpsertOutputFormat( new SimpleJdbcConnectionProvider( - JdbcConnectorOptions.builder() + InternalJdbcConnectionOptions.builder() .setDBUrl(getMetadata().getJdbcUrl()) .setTableName(OUTPUT_TABLE) .build()) { @@ -172,8 +172,8 @@ public class JdbcTableOutputFormatTest extends JdbcDataTestBase { @Test void testJdbcOutputFormat() throws Exception { - JdbcConnectorOptions options = - JdbcConnectorOptions.builder() + InternalJdbcConnectionOptions options = + InternalJdbcConnectionOptions.builder() .setDBUrl(getMetadata().getJdbcUrl()) .setTableName(OUTPUT_TABLE) .build(); diff --git a/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcAppendOnlyWriterTest.java b/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcAppendOnlyWriterTest.java index f0407dc..5e93b01 100644 --- a/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcAppendOnlyWriterTest.java +++ b/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcAppendOnlyWriterTest.java @@ -25,7 +25,7 @@ import org.apache.flink.connector.jdbc.JdbcTestBase; import org.apache.flink.connector.jdbc.databases.DatabaseMetadata; import org.apache.flink.connector.jdbc.dialect.JdbcDialectLoader; import org.apache.flink.connector.jdbc.internal.JdbcOutputFormat; -import org.apache.flink.connector.jdbc.internal.options.JdbcConnectorOptions; +import org.apache.flink.connector.jdbc.internal.options.InternalJdbcConnectionOptions; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; @@ -62,7 +62,7 @@ class JdbcAppendOnlyWriterTest extends JdbcTestBase { format = JdbcOutputFormat.builder() .setOptions( - JdbcConnectorOptions.builder() + InternalJdbcConnectionOptions.builder() .setDBUrl(getMetadata().getJdbcUrl()) .setDialect( JdbcDialectLoader.load( diff --git a/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableFactoryTest.java b/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableFactoryTest.java index 44d4d65..a23bf19 100644 --- a/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableFactoryTest.java +++ b/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableFactoryTest.java @@ -20,7 +20,7 @@ package org.apache.flink.connector.jdbc.table; import org.apache.flink.configuration.Configuration; import org.apache.flink.connector.jdbc.JdbcExecutionOptions; -import org.apache.flink.connector.jdbc.internal.options.JdbcConnectorOptions; +import org.apache.flink.connector.jdbc.internal.options.InternalJdbcConnectionOptions; import org.apache.flink.connector.jdbc.internal.options.JdbcDmlOptions; import org.apache.flink.connector.jdbc.internal.options.JdbcReadOptions; import org.apache.flink.table.api.DataTypes; @@ -72,8 +72,8 @@ class JdbcDynamicTableFactoryTest { // validation for source DynamicTableSource actualSource = createTableSource(SCHEMA, properties); - JdbcConnectorOptions options = - JdbcConnectorOptions.builder() + InternalJdbcConnectionOptions options = + InternalJdbcConnectionOptions.builder() .setDBUrl("jdbc:derby:memory:mydb") .setTableName("mytable") .setDriverName("org.apache.derby.jdbc.EmbeddedDriver") @@ -124,8 +124,8 @@ class JdbcDynamicTableFactoryTest { DynamicTableSource actual = createTableSource(SCHEMA, properties); - JdbcConnectorOptions options = - JdbcConnectorOptions.builder() + InternalJdbcConnectionOptions options = + InternalJdbcConnectionOptions.builder() .setDBUrl("jdbc:derby:memory:mydb") .setTableName("mytable") .build(); @@ -161,8 +161,8 @@ class JdbcDynamicTableFactoryTest { DynamicTableSource actual = createTableSource(SCHEMA, properties); - JdbcConnectorOptions options = - JdbcConnectorOptions.builder() + InternalJdbcConnectionOptions options = + InternalJdbcConnectionOptions.builder() .setDBUrl("jdbc:derby:memory:mydb") .setTableName("mytable") .build(); @@ -186,8 +186,8 @@ class JdbcDynamicTableFactoryTest { DynamicTableSource actual = createTableSource(SCHEMA, properties); - JdbcConnectorOptions options = - JdbcConnectorOptions.builder() + InternalJdbcConnectionOptions options = + InternalJdbcConnectionOptions.builder() .setDBUrl("jdbc:derby:memory:mydb") .setTableName("mytable") .build(); @@ -214,8 +214,8 @@ class JdbcDynamicTableFactoryTest { DynamicTableSink actual = createTableSink(SCHEMA, properties); - JdbcConnectorOptions options = - JdbcConnectorOptions.builder() + InternalJdbcConnectionOptions options = + InternalJdbcConnectionOptions.builder() .setDBUrl("jdbc:derby:memory:mydb") .setTableName("mytable") .build(); @@ -247,8 +247,8 @@ class JdbcDynamicTableFactoryTest { DynamicTableSink actual = createTableSink(SCHEMA, properties); - JdbcConnectorOptions options = - JdbcConnectorOptions.builder() + InternalJdbcConnectionOptions options = + InternalJdbcConnectionOptions.builder() .setDBUrl("jdbc:derby:memory:mydb") .setTableName("mytable") .setParallelism(2) @@ -371,8 +371,8 @@ class JdbcDynamicTableFactoryTest { DynamicTableSource actual = createTableSource(SCHEMA, properties); - JdbcConnectorOptions options = - JdbcConnectorOptions.builder() + InternalJdbcConnectionOptions options = + InternalJdbcConnectionOptions.builder() .setDBUrl("jdbc:derby:memory:mydb") .setTableName("mytable") .build(); diff --git a/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcOutputFormatTest.java b/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcOutputFormatTest.java index 897d485..dacacce 100644 --- a/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcOutputFormatTest.java +++ b/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcOutputFormatTest.java @@ -21,7 +21,7 @@ package org.apache.flink.connector.jdbc.table; import org.apache.flink.connector.jdbc.JdbcDataTestBase; import org.apache.flink.connector.jdbc.JdbcExecutionOptions; import org.apache.flink.connector.jdbc.internal.JdbcOutputFormat; -import org.apache.flink.connector.jdbc.internal.options.JdbcConnectorOptions; +import org.apache.flink.connector.jdbc.internal.options.InternalJdbcConnectionOptions; import org.apache.flink.connector.jdbc.internal.options.JdbcDmlOptions; import org.apache.flink.table.api.DataTypes; import org.apache.flink.table.data.RowData; @@ -89,8 +89,8 @@ class JdbcOutputFormatTest extends JdbcDataTestBase { String expectedMsg = "unable to open JDBC writer"; assertThatThrownBy( () -> { - JdbcConnectorOptions jdbcOptions = - JdbcConnectorOptions.builder() + InternalJdbcConnectionOptions jdbcOptions = + InternalJdbcConnectionOptions.builder() .setDriverName("org.apache.derby.jdbc.idontexist") .setDBUrl(DERBY_EBOOKSHOP_DB.getJdbcUrl()) .setTableName(INPUT_TABLE) @@ -120,8 +120,8 @@ class JdbcOutputFormatTest extends JdbcDataTestBase { void testInvalidURL() { assertThatThrownBy( () -> { - JdbcConnectorOptions jdbcOptions = - JdbcConnectorOptions.builder() + InternalJdbcConnectionOptions jdbcOptions = + InternalJdbcConnectionOptions.builder() .setDriverName(DERBY_EBOOKSHOP_DB.getDriverClass()) .setDBUrl("jdbc:der:iamanerror:mory:ebookshop") .setTableName(INPUT_TABLE) @@ -150,8 +150,8 @@ class JdbcOutputFormatTest extends JdbcDataTestBase { void testIncompatibleTypes() { assertThatThrownBy( () -> { - JdbcConnectorOptions jdbcOptions = - JdbcConnectorOptions.builder() + InternalJdbcConnectionOptions jdbcOptions = + InternalJdbcConnectionOptions.builder() .setDriverName(DERBY_EBOOKSHOP_DB.getDriverClass()) .setDBUrl(DERBY_EBOOKSHOP_DB.getJdbcUrl()) .setTableName(INPUT_TABLE) @@ -189,8 +189,8 @@ class JdbcOutputFormatTest extends JdbcDataTestBase { void testExceptionOnInvalidType() { assertThatThrownBy( () -> { - JdbcConnectorOptions jdbcOptions = - JdbcConnectorOptions.builder() + InternalJdbcConnectionOptions jdbcOptions = + InternalJdbcConnectionOptions.builder() .setDriverName(DERBY_EBOOKSHOP_DB.getDriverClass()) .setDBUrl(DERBY_EBOOKSHOP_DB.getJdbcUrl()) .setTableName(OUTPUT_TABLE) @@ -230,8 +230,8 @@ class JdbcOutputFormatTest extends JdbcDataTestBase { String expectedMsg = "Writing records to JDBC failed."; assertThatThrownBy( () -> { - JdbcConnectorOptions jdbcOptions = - JdbcConnectorOptions.builder() + InternalJdbcConnectionOptions jdbcOptions = + InternalJdbcConnectionOptions.builder() .setDriverName(DERBY_EBOOKSHOP_DB.getDriverClass()) .setDBUrl(DERBY_EBOOKSHOP_DB.getJdbcUrl()) .setTableName(OUTPUT_TABLE) @@ -276,8 +276,8 @@ class JdbcOutputFormatTest extends JdbcDataTestBase { @Test void testJdbcOutputFormat() throws IOException, SQLException { - JdbcConnectorOptions jdbcOptions = - JdbcConnectorOptions.builder() + InternalJdbcConnectionOptions jdbcOptions = + InternalJdbcConnectionOptions.builder() .setDriverName(DERBY_EBOOKSHOP_DB.getDriverClass()) .setDBUrl(DERBY_EBOOKSHOP_DB.getJdbcUrl()) .setTableName(OUTPUT_TABLE) @@ -329,8 +329,8 @@ class JdbcOutputFormatTest extends JdbcDataTestBase { @Test void testFlush() throws SQLException, IOException { - JdbcConnectorOptions jdbcOptions = - JdbcConnectorOptions.builder() + InternalJdbcConnectionOptions jdbcOptions = + InternalJdbcConnectionOptions.builder() .setDriverName(DERBY_EBOOKSHOP_DB.getDriverClass()) .setDBUrl(DERBY_EBOOKSHOP_DB.getJdbcUrl()) .setTableName(OUTPUT_TABLE_2) @@ -399,8 +399,8 @@ class JdbcOutputFormatTest extends JdbcDataTestBase { @Test void testFlushWithBatchSizeEqualsZero() throws SQLException, IOException { - JdbcConnectorOptions jdbcOptions = - JdbcConnectorOptions.builder() + InternalJdbcConnectionOptions jdbcOptions = + InternalJdbcConnectionOptions.builder() .setDriverName(DERBY_EBOOKSHOP_DB.getDriverClass()) .setDBUrl(DERBY_EBOOKSHOP_DB.getJdbcUrl()) .setTableName(OUTPUT_TABLE_2) @@ -446,8 +446,8 @@ class JdbcOutputFormatTest extends JdbcDataTestBase { @Test void testInvalidConnectionInJdbcOutputFormat() throws IOException, SQLException { - JdbcConnectorOptions jdbcOptions = - JdbcConnectorOptions.builder() + InternalJdbcConnectionOptions jdbcOptions = + InternalJdbcConnectionOptions.builder() .setDriverName(DERBY_EBOOKSHOP_DB.getDriverClass()) .setDBUrl(DERBY_EBOOKSHOP_DB.getJdbcUrl()) .setTableName(OUTPUT_TABLE_3) diff --git a/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcRowDataInputFormatTest.java b/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcRowDataInputFormatTest.java index 013e55e..a0040a2 100644 --- a/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcRowDataInputFormatTest.java +++ b/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcRowDataInputFormatTest.java @@ -21,7 +21,7 @@ package org.apache.flink.connector.jdbc.table; import org.apache.flink.connector.jdbc.JdbcDataTestBase; import org.apache.flink.connector.jdbc.JdbcTestFixture; import org.apache.flink.connector.jdbc.dialect.JdbcDialect; -import org.apache.flink.connector.jdbc.internal.options.JdbcConnectorOptions; +import org.apache.flink.connector.jdbc.internal.options.InternalJdbcConnectionOptions; import org.apache.flink.connector.jdbc.split.JdbcGenericParameterValuesProvider; import org.apache.flink.connector.jdbc.split.JdbcNumericBetweenParametersProvider; import org.apache.flink.connector.jdbc.split.JdbcParameterValuesProvider; @@ -66,7 +66,7 @@ class JdbcRowDataInputFormatTest extends JdbcDataTestBase { DataTypes.INT() }; final JdbcDialect dialect = - JdbcConnectorOptions.builder() + InternalJdbcConnectionOptions.builder() .setDBUrl(DERBY_EBOOKSHOP_DB.getJdbcUrl()) .setTableName(INPUT_TABLE) .build() diff --git a/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcRowDataLookupFunctionTest.java b/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcRowDataLookupFunctionTest.java index 72b083e..5759c52 100644 --- a/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcRowDataLookupFunctionTest.java +++ b/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcRowDataLookupFunctionTest.java @@ -18,7 +18,7 @@ package org.apache.flink.connector.jdbc.table; -import org.apache.flink.connector.jdbc.internal.options.JdbcConnectorOptions; +import org.apache.flink.connector.jdbc.internal.options.InternalJdbcConnectionOptions; import org.apache.flink.table.api.DataTypes; import org.apache.flink.table.connector.source.lookup.LookupOptions; import org.apache.flink.table.data.RowData; @@ -84,8 +84,8 @@ class JdbcRowDataLookupFunctionTest extends JdbcLookupTestBase { } private JdbcRowDataLookupFunction buildRowDataLookupFunction(boolean withFailure) { - JdbcConnectorOptions jdbcOptions = - JdbcConnectorOptions.builder() + InternalJdbcConnectionOptions jdbcOptions = + InternalJdbcConnectionOptions.builder() .setDriverName(DERBY_EBOOKSHOP_DB.getDriverClass()) .setDBUrl(DB_URL) .setTableName(LOOKUP_TABLE)
