wombatu-kun commented on code in PR #19117:
URL: https://github.com/apache/hudi/pull/19117#discussion_r3497939374
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/EventBuffers.java:
##########
@@ -171,7 +171,7 @@ public String getPendingInstants() {
*/
public Map<Long, Pair<String, EventBuffer>> getAllCompletedEvents() {
return this.eventBuffers.entrySet().stream()
- .filter(entry -> entry.getValue().getRight().allEventsCompleted())
+ .filter(entry -> entry.getValue().getRight().allEventsCompleted() &&
!entry.getValue().getRight().isEmptyDataWriteBuffer())
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
Review Comment:
The metric gauge calls `getAllCompletedEvents().size()` on every poll, so
this `Collectors.toMap` builds and discards a full map just to get a count.
Consider exposing a count-only path (for example `count()` over the same
filtered stream) so the reporting thread does not allocate a throwaway map each
interval.
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/EventBuffers.java:
##########
@@ -171,7 +171,7 @@ public String getPendingInstants() {
*/
public Map<Long, Pair<String, EventBuffer>> getAllCompletedEvents() {
Review Comment:
After adding the `!isEmptyDataWriteBuffer()` filter, this Javadoc ("...where
there exists no event sent by eager flushing from writers") is incomplete: the
method now also drops completed buffers whose data-write buffer is entirely
empty. Update it to mention the empty-data-write-buffer exclusion.
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java:
##########
@@ -507,6 +515,11 @@ private void initEventBufferIfNecessary() {
this.eventBuffers = EventBuffers.getInstance(conf, this.parallelism);
}
+ @VisibleForTesting
+ int getPendingCommitInstantCount() {
+ return this.eventBuffers == null ? 0 :
this.eventBuffers.getAllCompletedEvents().size();
Review Comment:
`getPendingCommitInstantCount()` runs on Flink's metric-reporter thread, but
`eventBuffers` is a plain field that the coordinator thread reassigns in
`initEventBufferIfNecessary` and nulls in `close()`. The null-check avoids an
NPE but not stale-reference visibility across threads. Consider making
`eventBuffers` volatile now that it is read off the coordinator thread.
##########
hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java:
##########
@@ -106,7 +106,7 @@ public class TestStreamWriteOperatorCoordinator {
@BeforeEach
public void before() throws Exception {
- coordinator =
createCoordinator(TestConfigurations.getDefaultConf(tempFile.getAbsolutePath()),
2);
+ coordinator =
createCoordinator(TestConfigurations.getDefaultConf("file://" +
tempFile.getAbsolutePath()), 2);
Review Comment:
This `file://` scheme is applied to every test in the class through
`before()`, but the change is unexplained and the pre-existing tests here
already drive `notifyCheckpointComplete` without a scheme. If the new test
needs an explicit scheme, scope the change to that test or add a short comment
on why it is required; otherwise it reads as unrelated.
--
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]
