Skip to content

Commit 118e644

Browse files
Samrat002Samrat002
committed
Serialization error issue
1 parent 1671a65 commit 118e644

15 files changed

+490
-285
lines changed

‎flink-connector-aws/flink-connector-redshift/src/main/java/org/apache/flink/connector/redshift/internal/connection/RedshiftJdbcConnectionProvider.java

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@
1818
package org.apache.flink.connector.redshift.internal.connection;
1919

2020
import org.apache.flink.annotation.Internal;
21-
import org.apache.flink.connector.redshift.options.InternalRedshiftConfigOptions;
21+
import org.apache.flink.connector.redshift.options.RedshiftConnectionConfigOptions;
2222

2323
import com.amazon.redshift.core.BaseConnection;
2424
import com.amazon.redshift.jdbc.RedshiftConnectionImpl;
@@ -44,10 +44,9 @@ public class RedshiftJdbcConnectionProvider implements RedshiftConnectionProvide
4444

4545
private transient BaseConnection connection;
4646

47-
private final transient InternalRedshiftConfigOptions.ConnectionConfigOptions connectionOptions;
47+
private final transient RedshiftConnectionConfigOptions connectionOptions;
4848

49-
public RedshiftJdbcConnectionProvider(
50-
InternalRedshiftConfigOptions.ConnectionConfigOptions configOptions) {
49+
public RedshiftJdbcConnectionProvider(RedshiftConnectionConfigOptions configOptions) {
5150
this.connectionOptions = configOptions;
5251
}
5352

‎flink-connector-aws/flink-connector-redshift/src/main/java/org/apache/flink/connector/redshift/internal/executor/RedshiftCopyModeBatchExecutor.java

Lines changed: 19 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,9 @@
2323
import org.apache.flink.connector.redshift.internal.statement.FieldNamedRedshiftPreparedStatement;
2424
import org.apache.flink.connector.redshift.internal.statement.RedshiftStatement;
2525
import org.apache.flink.connector.redshift.options.InternalRedshiftConfigOptions;
26+
import org.apache.flink.connector.redshift.options.RedshiftConnectionConfigOptions;
27+
import org.apache.flink.connector.redshift.options.RedshiftCopyModeConfigOptions;
28+
import org.apache.flink.connector.redshift.options.RedshiftExecutionConfigOptions;
2629
import org.apache.flink.connector.redshift.util.S3Util;
2730
import org.apache.flink.table.data.RowData;
2831
import org.apache.flink.table.types.logical.LogicalType;
@@ -43,19 +46,19 @@ public class RedshiftCopyModeBatchExecutor implements RedshiftExecutor {
4346

4447
private static final Logger LOG = LoggerFactory.getLogger(RedshiftCopyModeBatchExecutor.class);
4548

46-
private final transient int maxRetries;
49+
private final int maxRetries;
4750

48-
private final transient String tableName;
51+
private final String tableName;
4952

5053
private final transient String[] fieldNames;
5154

52-
private transient String sql;
55+
private String sql;
5356

54-
private final transient RedshiftCopyModeRowConverter copyRowConverter;
57+
private final RedshiftCopyModeRowConverter copyRowConverter;
5558

56-
private final transient String tempS3Uri;
59+
private final String tempS3Uri;
5760

58-
private final transient String iamRoleArn;
61+
private final String iamRoleArn;
5962

6063
private final transient List<String[]> csvData;
6164

@@ -64,16 +67,21 @@ public class RedshiftCopyModeBatchExecutor implements RedshiftExecutor {
6467
private transient FieldNamedRedshiftPreparedStatement statement;
6568

6669
public RedshiftCopyModeBatchExecutor(
67-
String[] fieldNames, LogicalType[] fieldTypes, InternalRedshiftConfigOptions options) {
68-
this.tableName = options.getConnectionConfigOptions().getTableName();
70+
String[] fieldNames,
71+
LogicalType[] fieldTypes,
72+
InternalRedshiftConfigOptions options,
73+
RedshiftConnectionConfigOptions connectionConfigOptions,
74+
RedshiftExecutionConfigOptions executionConfigOptions,
75+
RedshiftCopyModeConfigOptions copyModeConfigOptions) {
76+
this.tableName = connectionConfigOptions.getTableName();
6977
this.fieldNames = fieldNames;
70-
this.maxRetries = options.getExecutionConfigOptions().getMaxRetries();
78+
this.maxRetries = executionConfigOptions.getMaxRetries();
7179
this.csvData = new ArrayList<>();
7280
this.s3Client = S3Client.create();
7381
this.copyRowConverter = new RedshiftCopyModeRowConverter(fieldTypes);
7482

75-
this.tempS3Uri = S3Util.getS3UriWithFileName(options.getCopyModeConfigOptions().getS3Uri());
76-
this.iamRoleArn = options.getCopyModeConfigOptions().getIamRoleArn();
83+
this.tempS3Uri = S3Util.getS3UriWithFileName(copyModeConfigOptions.getS3Uri());
84+
this.iamRoleArn = copyModeConfigOptions.getIamRoleArn();
7785
}
7886

7987
@Override

‎flink-connector-aws/flink-connector-redshift/src/main/java/org/apache/flink/connector/redshift/internal/executor/RedshiftCopyModeUpsertExecutor.java

Lines changed: 19 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,9 @@
2222
import org.apache.flink.connector.redshift.internal.connection.RedshiftConnectionProvider;
2323
import org.apache.flink.connector.redshift.internal.statement.FieldNamedRedshiftPreparedStatement;
2424
import org.apache.flink.connector.redshift.internal.statement.RedshiftStatement;
25-
import org.apache.flink.connector.redshift.options.InternalRedshiftConfigOptions;
25+
import org.apache.flink.connector.redshift.options.RedshiftConnectionConfigOptions;
26+
import org.apache.flink.connector.redshift.options.RedshiftCopyModeConfigOptions;
27+
import org.apache.flink.connector.redshift.options.RedshiftExecutionConfigOptions;
2628
import org.apache.flink.connector.redshift.util.S3Util;
2729
import org.apache.flink.table.data.RowData;
2830
import org.apache.flink.table.types.logical.LogicalType;
@@ -45,29 +47,29 @@ public class RedshiftCopyModeUpsertExecutor implements RedshiftExecutor {
4547

4648
private static final Logger LOG = LoggerFactory.getLogger(RedshiftCopyModeUpsertExecutor.class);
4749

48-
private final transient int maxRetries;
50+
private final int maxRetries;
4951

50-
private final transient String tableName;
52+
private final String tableName;
5153

5254
private final transient String stageTableName;
5355

5456
private final transient String[] fieldNames;
5557

5658
private final transient String[] keyFields;
5759

58-
private final transient String tempS3Uri;
60+
private final String tempS3Uri;
5961

60-
private final transient String iamRoleArn;
62+
private final String iamRoleArn;
6163

62-
private transient String copyInsertSql;
64+
private String copyInsertSql;
6365

64-
private transient String updateTransactionSql;
66+
private String updateTransactionSql;
6567

66-
private transient String deleteSql;
68+
private String deleteSql;
6769

68-
private final transient RedshiftCopyModeRowConverter deleteConverter;
70+
private final RedshiftCopyModeRowConverter deleteConverter;
6971

70-
private final transient Function<RowData, RowData> deleteExtractor;
72+
private final Function<RowData, RowData> deleteExtractor;
7173

7274
private transient FieldNamedRedshiftPreparedStatement insertStatement;
7375

@@ -89,22 +91,24 @@ public RedshiftCopyModeUpsertExecutor(
8991
LogicalType[] fieldTypes,
9092
RedshiftCopyModeRowConverter deleteConverter,
9193
Function<RowData, RowData> deleteExtractor,
92-
InternalRedshiftConfigOptions options) {
94+
RedshiftConnectionConfigOptions connectionConfigOptions,
95+
RedshiftExecutionConfigOptions executionConfigOptions,
96+
RedshiftCopyModeConfigOptions copyModeConfigOptions) {
9397

94-
this.maxRetries = options.getExecutionConfigOptions().getMaxRetries();
98+
this.maxRetries = executionConfigOptions.getMaxRetries();
9599
this.fieldNames = fieldNames;
96100
this.keyFields = keyFields;
97101
this.deleteConverter = deleteConverter;
98102
this.deleteExtractor = deleteExtractor;
99103
this.csvInsertData = new ArrayList<>();
100104
this.csvUpdateData = new ArrayList<>();
101105

102-
this.tableName = options.getConnectionConfigOptions().getTableName();
103-
this.iamRoleArn = options.getCopyModeConfigOptions().getIamRoleArn();
106+
this.tableName = connectionConfigOptions.getTableName();
107+
this.iamRoleArn = copyModeConfigOptions.getIamRoleArn();
104108
this.s3Client = S3Client.create();
105109
this.copyRowConverter = new RedshiftCopyModeRowConverter(fieldTypes);
106110
this.stageTableName = "_" + tableName + "_stage";
107-
this.tempS3Uri = S3Util.getS3UriWithFileName(options.getCopyModeConfigOptions().getS3Uri());
111+
this.tempS3Uri = S3Util.getS3UriWithFileName(copyModeConfigOptions.getS3Uri());
108112
}
109113

110114
@Override

‎flink-connector-aws/flink-connector-redshift/src/main/java/org/apache/flink/connector/redshift/internal/executor/RedshiftExecutor.java

Lines changed: 54 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,9 @@
2525
import org.apache.flink.connector.redshift.internal.statement.FieldNamedRedshiftPreparedStatement;
2626
import org.apache.flink.connector.redshift.internal.statement.RedshiftStatement;
2727
import org.apache.flink.connector.redshift.options.InternalRedshiftConfigOptions;
28+
import org.apache.flink.connector.redshift.options.RedshiftConnectionConfigOptions;
29+
import org.apache.flink.connector.redshift.options.RedshiftCopyModeConfigOptions;
30+
import org.apache.flink.connector.redshift.options.RedshiftExecutionConfigOptions;
2831
import org.apache.flink.table.data.GenericRowData;
2932
import org.apache.flink.table.data.RowData;
3033
import org.apache.flink.table.types.logical.LogicalType;
@@ -96,6 +99,9 @@ default void attemptExecuteBatch(
9699

97100
static RedshiftExecutor createRedshiftExecutor(
98101
InternalRedshiftConfigOptions redshiftConfigOptions,
102+
RedshiftConnectionConfigOptions connectionConfigOptions,
103+
RedshiftExecutionConfigOptions executionConfigOptions,
104+
RedshiftCopyModeConfigOptions copyModeConfigOptions,
99105
String[] fieldNames,
100106
String[] primaryKeyFields,
101107
LogicalType[] fieldTypes) {
@@ -104,23 +110,40 @@ static RedshiftExecutor createRedshiftExecutor(
104110
case COPY:
105111
LOG.info("Create Upload Copy UPSERT Executor.");
106112
return createCopyModeUpsertExecutor(
107-
fieldNames, primaryKeyFields, fieldTypes, redshiftConfigOptions);
113+
fieldNames,
114+
primaryKeyFields,
115+
fieldTypes,
116+
connectionConfigOptions,
117+
executionConfigOptions,
118+
copyModeConfigOptions);
108119
case JDBC:
109120
LOG.info("Create pure JDBC UPSERT Executor.");
110121
return createJdbcModeUpsertExecutor(
111-
fieldNames, primaryKeyFields, fieldTypes, redshiftConfigOptions);
122+
fieldNames,
123+
primaryKeyFields,
124+
fieldTypes,
125+
connectionConfigOptions,
126+
executionConfigOptions);
112127
}
113128

114129
} else {
115130
switch (redshiftConfigOptions.getSinkMode()) {
116131
case JDBC:
117132
LOG.info("Create pure JDBC batch Executor.");
118133
return createJdbcModeBatchExecutor(
119-
fieldNames, fieldTypes, redshiftConfigOptions);
134+
fieldNames,
135+
fieldTypes,
136+
connectionConfigOptions,
137+
executionConfigOptions);
120138
case COPY:
121139
LOG.info("Create Upload Copy batch Executor.");
122140
return createCopyModeBatchExecutor(
123-
fieldNames, fieldTypes, redshiftConfigOptions);
141+
fieldNames,
142+
fieldTypes,
143+
redshiftConfigOptions,
144+
connectionConfigOptions,
145+
executionConfigOptions,
146+
copyModeConfigOptions);
124147
}
125148
}
126149
throw new FlinkRuntimeException("No correct Executor found");
@@ -129,15 +152,26 @@ static RedshiftExecutor createRedshiftExecutor(
129152
static RedshiftCopyModeBatchExecutor createCopyModeBatchExecutor(
130153
String[] fieldNames,
131154
LogicalType[] fieldTypes,
132-
InternalRedshiftConfigOptions redshiftConfigOptions) {
133-
return new RedshiftCopyModeBatchExecutor(fieldNames, fieldTypes, redshiftConfigOptions);
155+
InternalRedshiftConfigOptions redshiftConfigOptions,
156+
RedshiftConnectionConfigOptions connectionConfigOptions,
157+
RedshiftExecutionConfigOptions executionConfigOptions,
158+
RedshiftCopyModeConfigOptions copyModeConfigOptions) {
159+
return new RedshiftCopyModeBatchExecutor(
160+
fieldNames,
161+
fieldTypes,
162+
redshiftConfigOptions,
163+
connectionConfigOptions,
164+
executionConfigOptions,
165+
copyModeConfigOptions);
134166
}
135167

136168
static RedshiftCopyModeUpsertExecutor createCopyModeUpsertExecutor(
137169
String[] fieldNames,
138170
String[] keyFields,
139171
LogicalType[] fieldTypes,
140-
InternalRedshiftConfigOptions options) {
172+
RedshiftConnectionConfigOptions connectionConfigOptions,
173+
RedshiftExecutionConfigOptions executionConfigOptions,
174+
RedshiftCopyModeConfigOptions copyModeConfigOptions) {
141175

142176
int[] delFields =
143177
Arrays.stream(keyFields)
@@ -153,24 +187,30 @@ static RedshiftCopyModeUpsertExecutor createCopyModeUpsertExecutor(
153187
fieldTypes,
154188
new RedshiftCopyModeRowConverter(delTypes),
155189
createExtractor(fieldTypes, delFields),
156-
options);
190+
connectionConfigOptions,
191+
executionConfigOptions,
192+
copyModeConfigOptions);
157193
}
158194

159195
static RedshiftJdbcModeBatchExecutor createJdbcModeBatchExecutor(
160-
String[] fieldNames, LogicalType[] fieldTypes, InternalRedshiftConfigOptions options) {
196+
String[] fieldNames,
197+
LogicalType[] fieldTypes,
198+
RedshiftConnectionConfigOptions connectionConfigOptions,
199+
RedshiftExecutionConfigOptions executionConfigOptions) {
161200
String insertSql =
162201
RedshiftStatement.getInsertIntoStatement(
163-
options.getConnectionConfigOptions().getTableName(), fieldNames);
202+
connectionConfigOptions.getTableName(), fieldNames);
164203
RedshiftRowConverter converter = new RedshiftJdbcModeRowConverter(RowType.of(fieldTypes));
165-
return new RedshiftJdbcModeBatchExecutor(insertSql, converter, options);
204+
return new RedshiftJdbcModeBatchExecutor(insertSql, converter, executionConfigOptions);
166205
}
167206

168207
static RedshiftJdbcModeUpsertExecutor createJdbcModeUpsertExecutor(
169208
String[] fieldNames,
170209
String[] keyFields,
171210
LogicalType[] fieldTypes,
172-
InternalRedshiftConfigOptions options) {
173-
String tableName = options.getConnectionConfigOptions().getTableName();
211+
RedshiftConnectionConfigOptions connectionConfigOptions,
212+
RedshiftExecutionConfigOptions executionConfigOptions) {
213+
String tableName = connectionConfigOptions.getTableName();
174214
String insertSql = RedshiftStatement.getInsertIntoStatement(tableName, fieldNames);
175215
String updateSql = RedshiftStatement.getUpdateStatement(tableName, fieldNames, keyFields);
176216
String deleteSql = RedshiftStatement.getDeleteStatement(tableName, keyFields);
@@ -200,7 +240,7 @@ static RedshiftJdbcModeUpsertExecutor createJdbcModeUpsertExecutor(
200240
new RedshiftJdbcModeRowConverter(RowType.of(delTypes)),
201241
createExtractor(fieldTypes, updFields),
202242
createExtractor(fieldTypes, delFields),
203-
options,
243+
executionConfigOptions,
204244
fieldNames);
205245
}
206246

‎flink-connector-aws/flink-connector-redshift/src/main/java/org/apache/flink/connector/redshift/internal/executor/RedshiftJdbcModeBatchExecutor.java

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,7 @@
2121
import org.apache.flink.connector.redshift.converter.RedshiftRowConverter;
2222
import org.apache.flink.connector.redshift.internal.connection.RedshiftConnectionProvider;
2323
import org.apache.flink.connector.redshift.internal.statement.FieldNamedRedshiftPreparedStatement;
24-
import org.apache.flink.connector.redshift.options.InternalRedshiftConfigOptions;
24+
import org.apache.flink.connector.redshift.options.RedshiftExecutionConfigOptions;
2525
import org.apache.flink.table.data.RowData;
2626
import org.apache.flink.util.FlinkRuntimeException;
2727

@@ -37,19 +37,21 @@ public class RedshiftJdbcModeBatchExecutor implements RedshiftExecutor {
3737

3838
private static final Logger LOG = LoggerFactory.getLogger(RedshiftJdbcModeBatchExecutor.class);
3939

40-
private final transient String sql;
40+
private final String sql;
4141

4242
private final transient RedshiftRowConverter converter;
4343

44-
private final transient int maxRetries;
44+
private final int maxRetries;
4545

4646
private transient FieldNamedRedshiftPreparedStatement statement;
4747

4848
public RedshiftJdbcModeBatchExecutor(
49-
String sql, RedshiftRowConverter converter, InternalRedshiftConfigOptions options) {
49+
String sql,
50+
RedshiftRowConverter converter,
51+
RedshiftExecutionConfigOptions executionConfigOptions) {
5052
this.sql = sql;
5153
this.converter = converter;
52-
this.maxRetries = options.getExecutionConfigOptions().getMaxRetries();
54+
this.maxRetries = executionConfigOptions.getMaxRetries();
5355
}
5456

5557
@Override

‎flink-connector-aws/flink-connector-redshift/src/main/java/org/apache/flink/connector/redshift/internal/executor/RedshiftJdbcModeUpsertExecutor.java

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,7 @@
2121
import org.apache.flink.connector.redshift.converter.RedshiftRowConverter;
2222
import org.apache.flink.connector.redshift.internal.connection.RedshiftConnectionProvider;
2323
import org.apache.flink.connector.redshift.internal.statement.FieldNamedRedshiftPreparedStatement;
24-
import org.apache.flink.connector.redshift.options.InternalRedshiftConfigOptions;
24+
import org.apache.flink.connector.redshift.options.RedshiftExecutionConfigOptions;
2525
import org.apache.flink.table.data.RowData;
2626
import org.apache.flink.util.FlinkRuntimeException;
2727

@@ -39,11 +39,11 @@ public class RedshiftJdbcModeUpsertExecutor implements RedshiftExecutor {
3939

4040
private static final Logger LOG = LoggerFactory.getLogger(RedshiftJdbcModeUpsertExecutor.class);
4141

42-
private final transient String insertSql;
42+
private final String insertSql;
4343

44-
private final transient String updateSql;
44+
private final String updateSql;
4545

46-
private final transient String deleteSql;
46+
private final String deleteSql;
4747

4848
private final transient RedshiftRowConverter insertConverter;
4949

@@ -55,7 +55,7 @@ public class RedshiftJdbcModeUpsertExecutor implements RedshiftExecutor {
5555

5656
private final transient Function<RowData, RowData> deleteExtractor;
5757

58-
private final transient int maxRetries;
58+
private final int maxRetries;
5959
private final transient String[] fieldNames;
6060

6161
private transient FieldNamedRedshiftPreparedStatement insertStatement;
@@ -73,7 +73,7 @@ public RedshiftJdbcModeUpsertExecutor(
7373
RedshiftRowConverter deleteConverter,
7474
Function<RowData, RowData> updateExtractor,
7575
Function<RowData, RowData> deleteExtractor,
76-
InternalRedshiftConfigOptions options,
76+
RedshiftExecutionConfigOptions executionConfigOptions,
7777
String[] fieldNames) {
7878
this.insertSql = insertSql;
7979
this.updateSql = updateSql;
@@ -83,7 +83,7 @@ public RedshiftJdbcModeUpsertExecutor(
8383
this.deleteConverter = deleteConverter;
8484
this.updateExtractor = updateExtractor;
8585
this.deleteExtractor = deleteExtractor;
86-
this.maxRetries = options.getExecutionConfigOptions().getMaxRetries();
86+
this.maxRetries = executionConfigOptions.getMaxRetries();
8787
this.fieldNames = fieldNames;
8888
}
8989

0 commit comments

Comments
 (0)