danny0405 commented on code in PR #17838:
URL: https://github.com/apache/hudi/pull/17838#discussion_r2706741577
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java:
##########
@@ -573,40 +588,54 @@ private boolean commitInstants(long checkpointId) {
*
* @return true if the write statuses are committed successfully.
*/
- private boolean commitInstant(long checkpointId, String instant,
WriteMetadataEvent[] eventBuffer) {
- if (Arrays.stream(eventBuffer).allMatch(Objects::isNull)) {
- // all the tasks are reset by failover, reset the while buffer and
returns early.
+ private boolean commitInstant(long checkpointId, String instant,
Pair<WriteMetadataEvent[], WriteMetadataEvent[]> eventBufferPair) {
+ if (Arrays.stream(eventBufferPair.getLeft()).allMatch(Objects::isNull)) {
+ // all the data write tasks are reset by failover, reset the while
buffer and returns early.
this.eventBuffers.reset(checkpointId);
// stop the heart beat for lazy cleaning
writeClient.getHeartbeatClient().stop(instant);
+ // stop the heartbeat for metadata table instant if streaming index
write is enabled
+ if (isStreamingIndexWriteEnabled) {
+ writeClient.stopCommitForMetadataTable(instant);
+ }
return false;
}
- List<WriteStatus> writeResults = Arrays.stream(eventBuffer)
+ List<WriteStatus> dataWriteResults =
Arrays.stream(eventBufferPair.getLeft())
.filter(Objects::nonNull)
.map(WriteMetadataEvent::getWriteStatuses)
.flatMap(Collection::stream)
.collect(Collectors.toList());
- if (writeResults.isEmpty() &&
!OptionsResolver.allowCommitOnEmptyBatch(conf)) {
+ List<WriteStatus> indexWriteResults =
Arrays.stream(eventBufferPair.getRight())
+ .filter(Objects::nonNull)
+ .map(WriteMetadataEvent::getWriteStatuses)
+ .flatMap(Collection::stream)
+ .collect(Collectors.toList());
+
+ if (dataWriteResults.isEmpty() &&
!OptionsResolver.allowCommitOnEmptyBatch(conf)) {
// No data has written, reset the buffer and returns early
this.eventBuffers.reset(checkpointId);
// stop the heart beat for lazy cleaning
writeClient.getHeartbeatClient().stop(instant);
+ // stop the heartbeat for metadata table instant if streaming index
write is enabled
Review Comment:
add a new method `writeClient.stopHeartbeat(String instant)` to cover both
the data and MDT hearbeats.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
