AHeise commented on code in PR #25292:
URL: https://github.com/apache/flink/pull/25292#discussion_r1758844552
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactorOperator.java:
##########
@@ -138,15 +136,15 @@ public void processElement(StreamRecord<CompactorRequest>
element) throws Except
@Override
public void endInput() throws Exception {
// add collecting requests into the final snapshot
- checkpointRequests.put(Long.MAX_VALUE, collectingRequests);
+ checkpointRequests.put(CommittableMessage.EOI, collectingRequests);
collectingRequests = new ArrayList<>();
// submit all requests and wait until they are done
- submitUntil(Long.MAX_VALUE);
+ submitUntil(CommittableMessage.EOI);
assert checkpointRequests.isEmpty();
getAllTasksFuture().join();
- emitCompacted(null);
Review Comment:
Afaik yes. We pass the param directly to both `CommittableMessage`s. The
serializer replaces `null` with `EOI`. So it effectively results in the same
bytes.
--
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
