ruanhang1993 commented on code in PR #4178:
URL: https://github.com/apache/flink-cdc/pull/4178#discussion_r3773203751
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlRecordEmitter.java:
##########
@@ -121,15 +125,33 @@ private void emitElement(SourceRecord element,
SourceOutput<T> output) throws Ex
debeziumDeserializationSchema.deserialize(element, outputCollector);
}
- public void applySplit(MySqlSplit split) {}
Review Comment:
[MySqlSourceReader.java:115](https://github.com/apache/flink-cdc/blob/91040dece074c18640a58f74a6391cf7613f8063/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlSourceReader.java#L115)
still use this method.
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/metrics/MySqlSourceReaderMetrics.java:
##########
@@ -51,4 +74,34 @@ public long getFetchDelay() {
public void recordFetchDelay(long fetchDelay) {
this.fetchDelay = fetchDelay;
}
+
+ public void numRecordsOutByDataChangeRecord(TableId tableId, OperationType
op) {
+ Tuple2<TableId, OperationType> metricMapKey = new Tuple2<>(tableId,
op);
+
+ Counter counter =
+ numRecordsOutByDataChangeRecordMap.compute(
Review Comment:
These metrics are registered lazily per (table, operation) pair. As a
result, a table that has only INSERT events will not expose the UPDATE or
DELETE metrics at all, rather than exposing them with a value of zero.
##########
docs/content/docs/connectors/flink-sources/mysql-cdc.md:
##########
@@ -941,6 +941,14 @@ Notice:
1. The group name is `namespace.schema.table`, where `namespace` is the actual
database name, `schema` is the actual schema name, and `table` is the actual
table name.
2. For MySQL, the `namespace` will be set to the default value "", and the
group name will be like `test_database.test_table`.
+The mysql-cdc connector offers six additional metrics for each type of data
change record.
+- `numRecordsOutByDataChangeRecordInsert`: The number of `INSERT` data change
records.
Review Comment:
The metric names registered do not match the names documented in
mysql-cdc.md.
For example，the metric name in the code is
`numRecordsOutDataChangeRecordInsert`.
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/metrics/MySqlSourceReaderMetrics.java:
##########
@@ -17,15 +17,33 @@
package org.apache.flink.cdc.connectors.mysql.source.metrics;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.cdc.common.event.OperationType;
import org.apache.flink.cdc.connectors.mysql.source.reader.MySqlSourceReader;
+import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.Gauge;
+import org.apache.flink.metrics.Meter;
+import org.apache.flink.metrics.MeterView;
import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.runtime.metrics.MetricNames;
+import io.debezium.relational.TableId;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
/** A collection class for handling metrics in {@link MySqlSourceReader}. */
public class MySqlSourceReaderMetrics {
public static final long UNDEFINED = -1;
+ private static final Map<OperationType, String> DATA_CHANGE_RECORD_MAP =
+ new ConcurrentHashMap<OperationType, String>() {
+ {
+ put(OperationType.INSERT, "DataChangeRecordInsert");
+ put(OperationType.UPDATE, "DataChangeRecordUpdate");
+ put(OperationType.DELETE, "DataChangeRecordDelete");
+ }
+ };
Review Comment:
Since this PR introduces a new externally observable metrics contract, could
we add tests covering them？
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlRecordEmitter.java:
##########
@@ -121,15 +125,33 @@ private void emitElement(SourceRecord element,
SourceOutput<T> output) throws Ex
debeziumDeserializationSchema.deserialize(element, outputCollector);
}
- public void applySplit(MySqlSplit split) {}
-
- private void reportMetrics(SourceRecord element) {
+ private void reportMetrics(SourceRecord record) {
+ Struct value = (Struct) record.value();
+ if (value != null) {
+ TableId tableId = RecordUtils.getTableId(record);
+ Envelope.Operation op =
+
Envelope.Operation.forCode(value.getString(Envelope.FieldName.OPERATION));
+ switch (op) {
+ case CREATE:
Review Comment:
Snapshot records use Envelope.Operation.READ, but
`RowDataDebeziumDeserializeSchema` emits them as `RowKind.INSERT`. Should READ
also increment the INSERT metric?
Are these metrics intended to cover only incremental binlog records? If so,
could we document explicitly that snapshot READ records are excluded?
--
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]
