You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "cshuo (via GitHub)" <gi...@apache.org> on 2026/06/30 07:15:43 UTC
[PR] feat(flink): Add pendingCommitInstantCount metric for write coordinator [hudi]
cshuo opened a new pull request, #19117:
URL: https://github.com/apache/hudi/pull/19117
### Describe the issue this Pull Request addresses
Flink stream writes currently do not expose coordinator-side visibility into how many write instants are waiting to be committed. This PR adds a write coordinator metric for pending commit instants and wires it directly to the coordinator's completed event buffers.
### Summary and Changelog
- Added `FlinkWriteCoordinatorMetrics` to register the `pendingCommitInstantCount` gauge.
- Wired `StreamWriteOperatorCoordinator` to register coordinator metrics on start and expose the pending commit instant count for testing.
- Updated `EventBuffers#getAllCompletedEvents` to exclude empty data-write buffers from the completed-events count.
- Added coverage in `TestFlinkWriteCoordinatorMetrics` for gauge registration and dynamic gauge values.
- Added coordinator-level coverage in `TestStreamWriteOperatorCoordinator` to verify pending commit count before and after checkpoint completion.
### Impact
Adds a new Flink write coordinator metric, `pendingCommitInstantCount`, for observing completed instants pending commit.
### Risk Level
low.
### Documentation Update
Not included in this commit. If Flink write coordinator metrics are documented in a metrics reference, `pendingCommitInstantCount` should be added there.
### Contributor's checklist
- [ ] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute)
- [ ] Enough context is provided in the sections above
- [ ] Adequate tests were added if applicable
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(flink): Add pendingCommitInstantCount metric for write coordinator [hudi]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(flink): Add pendingCommitInstantCount metric for write coordinator [hudi]
Posted by "hudi-agent (via GitHub)" <gi...@apache.org>.
hudi-agent commented on code in PR #19117:
URL: https://github.com/apache/hudi/pull/19117#discussion_r3496905208
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/metrics/FlinkWriteCoordinatorMetrics.java:
##########
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.metrics;
+
+import org.apache.flink.metrics.MetricGroup;
+
+import java.util.function.IntSupplier;
+
+/**
+ * Metrics for Flink stream write operator coordinator.
+ */
+public class FlinkWriteCoordinatorMetrics extends HoodieFlinkMetrics {
+
+ private static final String PENDING_COMMIT_INSTANT_COUNT = "pendingCommitInstantCount";
Review Comment:
🤖 nit: the parent class is `HoodieFlinkMetrics`, so this subclass would be more consistent with the project's `Hoodie`-prefix convention if it were named `HoodieFlinkWriteCoordinatorMetrics`.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub>
##########
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()
Review Comment:
🤖 nit: `entry.getValue().getRight()` is traversed twice in the same predicate — could you extract it to a local variable (e.g. `EventBuffer buf = entry.getValue().getRight()`) before the stream, or use a helper in the lambda, to make the two conditions easier to read?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub>
##########
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())
Review Comment:
🤖 `getAllCompletedEvents()` isn't only used for the new metric — it's also the source serialized into coordinator checkpoint state in `checkpointCoordinator()` (StreamWriteOperatorCoordinator line 301). Adding `!isEmptyDataWriteBuffer()` here means empty-data-write buffers are now excluded from what gets persisted and restored on `resetToCheckpoint`. Is that intended? It looks consistent with `commitInstant`'s early-return for empty buffers, but on recovery those instants would no longer be restored at all, so the `reset(...)` + `writeClient.cleanResources(instant)` cleanup in `commitInstant` wouldn't run for them. Could you confirm recovery semantics are unaffected, or alternatively give the metric its own helper so the checkpoint path keeps its current behavior?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(flink): Add pendingCommitInstantCount metric for write coordinator [hudi]
Posted by "voonhous (via GitHub) via commits" <co...@hudi.apache.org>.
voonhous commented on PR #19117:
URL: https://github.com/apache/hudi/pull/19117#issuecomment-6701153199
Temporarily closing this PR while we debug a GitHub settings error on master (#20471); it will be reopened shortly, no action needed.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(flink): Add pendingCommitInstantCount metric for write coordinator [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19117:
URL: https://github.com/apache/hudi/pull/19117#issuecomment-4840900211
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dca19c82a25e91e5995446428667540720ba7e2a",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "dca19c82a25e91e5995446428667540720ba7e2a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* dca19c82a25e91e5995446428667540720ba7e2a UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(flink): Add pendingCommitInstantCount metric for write coordinator [hudi]
Posted by "voonhous (via GitHub) via commits" <co...@hudi.apache.org>.
voonhous closed pull request #19117: feat(flink): Add pendingCommitInstantCount metric for write coordinator
URL: https://github.com/apache/hudi/pull/19117
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(flink): Add pendingCommitInstantCount metric for write coordinator [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19117:
URL: https://github.com/apache/hudi/pull/19117#issuecomment-4841035139
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dca19c82a25e91e5995446428667540720ba7e2a",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15037",
"triggerID" : "dca19c82a25e91e5995446428667540720ba7e2a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* dca19c82a25e91e5995446428667540720ba7e2a Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15037)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(flink): Add pendingCommitInstantCount metric for write coordinator [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19117:
URL: https://github.com/apache/hudi/pull/19117#issuecomment-4840934177
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dca19c82a25e91e5995446428667540720ba7e2a",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15037",
"triggerID" : "dca19c82a25e91e5995446428667540720ba7e2a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* dca19c82a25e91e5995446428667540720ba7e2a Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15037)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(flink): Add pendingCommitInstantCount metric for write coordinator [hudi]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
wombatu-kun commented on code in PR #19117:
URL: https://github.com/apache/hudi/pull/19117#discussion_r3497887193
##########
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())
Review Comment:
No test exercises this new `!isEmptyDataWriteBuffer()` clause. Both `testPendingCommitInstantCountGauge` and `TestStreamWriteOperatorCoordinator#testPendingCommitInstantCount` only add non-null data-write events, so `isEmptyDataWriteBuffer()` is always false and deleting this clause would not fail either test. Add a case that builds a completed buffer whose data-write buffer is all null (for example only metadata/index events, or a failover-reset buffer) and asserts it is excluded from the count.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org