dawidwys commented on code in PR #23601:
URL: https://github.com/apache/flink/pull/23601#discussion_r1373031168
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesRuntimeFunctions.java:
##########
@@ -283,22 +293,29 @@ private abstract static class AbstractExactlyOnceSink
extends RichSinkFunction<R
private static final long serialVersionUID = 1L;
protected final String tableName;
+ protected final DataType consumedDataType;
+ protected final DataStructureConverter converter;
+ protected transient ListState<Row> rawResultState;
+ protected transient List<Row> localRawResult;
- protected transient ListState<String> rawResultState;
- protected transient List<String> localRawResult;
-
- protected AbstractExactlyOnceSink(String tableName) {
+ protected AbstractExactlyOnceSink(
+ String tableName, DataType consumedDataType,
DataStructureConverter converter) {
this.tableName = tableName;
+ this.consumedDataType = consumedDataType;
+ this.converter = converter;
}
@Override
public void initializeState(FunctionInitializationContext context)
throws Exception {
this.rawResultState =
context.getOperatorStateStore()
- .getListState(new
ListStateDescriptor<>("sink-results", Types.STRING));
+ .getListState(
+ new ListStateDescriptor<>(
+ "sink-results",
+
ExternalSerializer.of(consumedDataType, true)));
Review Comment:
Maybe no tests, depend on this particular one.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
