JingsongLi commented on a change in pull request #1185:
URL: https://github.com/apache/iceberg/pull/1185#discussion_r478798549
##########
File path:
flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriter.java
##########
@@ -34,15 +35,16 @@
private static final long serialVersionUID = 1L;
private final String fullTableName;
+ private final TaskWriterFactory<T> taskWriterFactory;
- private transient TaskWriterFactory<T> taskWriterFactory;
private transient TaskWriter<T> writer;
private transient int subTaskId;
private transient int attemptId;
IcebergStreamWriter(String fullTableName, TaskWriterFactory<T>
taskWriterFactory) {
this.fullTableName = fullTableName;
this.taskWriterFactory = taskWriterFactory;
+ setChainingStrategy(ChainingStrategy.ALWAYS);
Review comment:
The chaining strategy definition:
```
* Defines the chaining scheme for the operator. When an operator is chained
to the
* predecessor, it means that they run in the same thread. They become one
operator
* consisting of multiple steps.
*
* <p>The default value used by the StreamOperator is {@link #HEAD}, which
means that
* the operator is not chained to its predecessor. Most operators override
this with
* {@link #ALWAYS}, meaning they will be chained to predecessors whenever
possible.
```
We should also set the chaining strategy of the writer, so as to avoid the
network crossing, which will bring additional serialization cost.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
