kbendick commented on a change in pull request #2109:
URL: https://github.com/apache/iceberg/pull/2109#discussion_r559870969
##########
File path:
flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java
##########
@@ -126,9 +137,16 @@ public void initializeState(StateInitializationContext
context) throws Exception
this.manifestOutputFileFactory =
FlinkManifestUtil.createOutputFileFactory(table, flinkJobId, subTaskId,
attemptId);
this.maxCommittedCheckpointId = INITIAL_CHECKPOINT_ID;
+ Map<String, String> properties = this.table.properties();
+ this.currentWatermark = PropertyUtil.propertyAsLong(properties,
CURRENT_WATERMARK, -1L);
+ this.storeWatermark = PropertyUtil.propertyAsBoolean(properties,
STORE_WATERMARK, false);
+
this.checkpointsState =
context.getOperatorStateStore().getListState(STATE_DESCRIPTOR);
this.jobIdState =
context.getOperatorStateStore().getListState(JOB_ID_DESCRIPTOR);
+ this.watermarkState =
context.getOperatorStateStore().getListState(WATERMARK_DESCRIPTOR);
+
if (context.isRestored()) {
+ watermarkPerCheckpoint.putAll(watermarkState.get().iterator().next());
Review comment:
Actually, on further inspection of the `ListState` interface, it says
that passing `null` to `putAll` is a no-op. So I don' think there should be
backwards compatibility issues, but should we possibly be (1) logging something
if no watermark state is restored and/or (2) possibly doing our own `null`
check vs relying on the documented behavior of `ListState#putAll` to be
consistent over time when inserting `null`?
----------------------------------------------------------------
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]
