[
https://issues.apache.org/jira/browse/FLINK-41201?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-41201:
-----------------------------------
Labels: pull-request-available (was: )
> Fluss pipeline sink error counters remain zero after write failures
> -------------------------------------------------------------------
>
> Key: FLINK-41201
> URL: https://issues.apache.org/jira/browse/FLINK-41201
> Project: Flink
> Issue Type: Bug
> Components: Flink CDC
> Reporter: Xiaobing Fang
> Priority: Major
> Labels: pull-request-available
>
> h2. Description
> The Fluss pipeline sink updates unregistered error counters after an
> asynchronous write failure. The registered numRecordsOutErrors and
> numRecordsSendErrors counters remain zero, so metric reporters do not observe
> the error count even though the writer propagates the failure.
> Both FlussSink.createWriter overloads call
> InternalSinkWriterMetricGroup.wrap(context.metricGroup()). However, Flink has
> already created the sink writer metric group before passing it to the
> connector.
> The additional wrap attempts to register four counters again. Flink retains
> the previously registered metrics on name collisions, but the new wrapper
> holds a different error counter. FlussSinkWriter.checkAsyncException()
> increments this unregistered counter.
> On the verified Flink versions, numRecordsSend and numBytesSend reuse the
> underlying operator IO counters and do not suffer from this counter
> separation.
> h2. Steps to reproduce
> # Create a FlussSink writer with a framework-provided SinkWriterMetricGroup.
> # Submit a record targeting a nonexistent Fluss table to trigger a write
> failure.
> # Execute the queued mailbox error callback or call flush().
> # Inspect the counters captured through MetricRegistry.register().
> The failure is propagated, but the registered numRecordsOutErrors and
> numRecordsSendErrors counters remain at 0 instead of increasing to 1.
> The regression reproducer covers both createWriter overloads and both error
> propagation paths.
> h2. Expected behavior and proposed fix
> Pass context.metricGroup() directly to FlussSinkWriter.initialize() in both
> createWriter overloads.
> The writer should update the same counters that Flink registered with metric
> reporters. This removes duplicate registration and restores error metric
> reporting without changing write behavior, exception propagation, or
> checkpoint semantics.
> h2. Validation
> Before the fix, all six new regression cases fail on the expected metric
> assertions.
> After the fix, the regression tests pass with Flink 1.20.3 on Java 11 and
> Java 17, and with Flink 2.2.0 on Java 17.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
