Xiaobing Fang created FLINK-41201:
-------------------------------------
Summary: 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
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)
