Skip to content

Commit dfe6e1f

Browse files
HZY-WadeHZY-Wade
committed
[connector/flink] Report pendingRecords.
this closes #138
1 parent 72f1e0c commit dfe6e1f

9 files changed

Lines changed: 206 additions & 1 deletion

File tree

‎fluss-client/src/main/java/com/alibaba/fluss/client/metrics/ScannerMetricGroup.java‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,13 +23,17 @@
2323
import com.alibaba.fluss.metrics.CharacterFilter;
2424
import com.alibaba.fluss.metrics.Counter;
2525
import com.alibaba.fluss.metrics.DescriptiveStatisticsHistogram;
26+
import com.alibaba.fluss.metrics.Gauge;
2627
import com.alibaba.fluss.metrics.Histogram;
2728
import com.alibaba.fluss.metrics.MeterView;
29+
import com.alibaba.fluss.metrics.Metric;
2830
import com.alibaba.fluss.metrics.MetricNames;
2931
import com.alibaba.fluss.metrics.ThreadSafeSimpleCounter;
3032
import com.alibaba.fluss.metrics.groups.AbstractMetricGroup;
3133
import com.alibaba.fluss.rpc.metrics.ClientMetricGroup;
34+
import com.alibaba.fluss.server.metrics.group.BucketMetricGroup;
3235

36+
import java.util.HashMap;
3337
import java.util.Map;
3438

3539
import static com.alibaba.fluss.metrics.utils.MetricGroupUtils.makeScope;
@@ -55,6 +59,7 @@ public class ScannerMetricGroup extends AbstractMetricGroup {
5559
private volatile double pollIdleRatio;
5660
private volatile long lastPollMs;
5761
private volatile long pollStartMs;
62+
private final Map<Integer, BucketMetricGroup> buckets = new HashMap<>();
5863

5964
public ScannerMetricGroup(ClientMetricGroup parent, TablePath tablePath) {
6065
super(parent.getMetricRegistry(), makeScope(parent, name), parent);
@@ -120,6 +125,18 @@ private long lastPollSecondsAgo() {
120125
return (System.currentTimeMillis() - lastPollMs) / 1000;
121126
}
122127

128+
public void recordBucketLag(int bucketId, long lag) {
129+
buckets.computeIfAbsent(
130+
bucketId, (bucket) -> new BucketMetricGroup(registry, bucketId, this));
131+
Metric metric = buckets.get(bucketId).metrics().get(bucketRecordsLagMetricName(bucketId));
132+
if (metric == null) {
133+
SimpleGauge simpleGauge = new SimpleGauge(lag);
134+
buckets.get(bucketId).gauge(bucketRecordsLagMetricName(bucketId), simpleGauge);
135+
} else {
136+
((SimpleGauge) metric).setValue(lag);
137+
}
138+
}
139+
123140
@Override
124141
protected String getGroupName(CharacterFilter filter) {
125142
return name;
@@ -130,4 +147,32 @@ protected final void putVariables(Map<String, String> variables) {
130147
variables.put("database", tablePath.getDatabaseName());
131148
variables.put("table", tablePath.getTableName());
132149
}
150+
151+
private static String bucketRecordsLagMetricName(int bucketId) {
152+
return bucketId + ".records-lag";
153+
}
154+
155+
@Override
156+
public Map<String, Metric> metrics() {
157+
Map<String, Metric> allMetric = new HashMap<>(super.metrics());
158+
buckets.forEach((key, value) -> allMetric.putAll(value.metrics()));
159+
return allMetric;
160+
}
161+
162+
private static class SimpleGauge implements Gauge<Long> {
163+
private long value;
164+
165+
public SimpleGauge(long value) {
166+
this.value = value;
167+
}
168+
169+
@Override
170+
public Long getValue() {
171+
return value;
172+
}
173+
174+
public void setValue(long value) {
175+
this.value = value;
176+
}
177+
}
133178
}

‎fluss-client/src/main/java/com/alibaba/fluss/client/scanner/log/FlussLogScanner.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
import com.alibaba.fluss.metadata.TableBucket;
2727
import com.alibaba.fluss.metadata.TableInfo;
2828
import com.alibaba.fluss.metadata.TablePath;
29+
import com.alibaba.fluss.metrics.Metric;
2930
import com.alibaba.fluss.rpc.RpcClient;
3031
import com.alibaba.fluss.rpc.metrics.ClientMetricGroup;
3132
import com.alibaba.fluss.types.RowType;
@@ -304,4 +305,9 @@ public void close() {
304305
release();
305306
}
306307
}
308+
309+
@Override
310+
public Map<String, Metric> metrics() {
311+
return Collections.unmodifiableMap(scannerMetricGroup.metrics());
312+
}
307313
}

‎fluss-client/src/main/java/com/alibaba/fluss/client/scanner/log/LogFetcher.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -314,6 +314,9 @@ private synchronized void handleFetchLogResponse(
314314
projection);
315315
logFetchBuffer.add(completedFetch);
316316
}
317+
scannerMetricGroup.recordBucketLag(
318+
tb.getBucket(),
319+
fetchResultForBucket.getHighWatermark() - fetchOffset);
317320
}
318321
}
319322
}

‎fluss-client/src/main/java/com/alibaba/fluss/client/scanner/log/LogScanner.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,8 +17,10 @@
1717
package com.alibaba.fluss.client.scanner.log;
1818

1919
import com.alibaba.fluss.annotation.PublicEvolving;
20+
import com.alibaba.fluss.metrics.Metric;
2021

2122
import java.time.Duration;
23+
import java.util.Map;
2224

2325
/**
2426
* The scanner is used to scan log data of specify table from Fluss.
@@ -120,4 +122,7 @@ default void subscribeFromBeginning(long partitionId, int bucket) {
120122
* #poll(Duration timeout)}.
121123
*/
122124
void wakeup();
125+
126+
/** Get the metrics of the log scanner. */
127+
public Map<String, ? extends Metric> metrics();
123128
}

‎fluss-common/src/main/java/com/alibaba/fluss/metrics/groups/AbstractMetricGroup.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -371,4 +371,8 @@ protected enum ChildType {
371371
VALUE,
372372
GENERIC
373373
}
374+
375+
public Map<String, Metric> metrics() {
376+
return this.metrics;
377+
}
374378
}

‎fluss-connectors/fluss-connector-flink/src/main/java/com/alibaba/fluss/connector/flink/source/metrics/FlinkSourceReaderMetrics.java‎

Lines changed: 58 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,15 +18,21 @@
1818

1919
import com.alibaba.fluss.connector.flink.source.reader.FlinkSourceReader;
2020
import com.alibaba.fluss.metadata.TableBucket;
21+
import com.alibaba.fluss.metrics.Gauge;
22+
import com.alibaba.fluss.metrics.Metric;
2123

2224
import org.apache.flink.metrics.MetricGroup;
2325
import org.apache.flink.metrics.groups.SourceReaderMetricGroup;
2426
import org.apache.flink.runtime.metrics.MetricNames;
2527
import org.slf4j.Logger;
2628
import org.slf4j.LoggerFactory;
2729

30+
import javax.annotation.Nullable;
31+
2832
import java.util.HashMap;
2933
import java.util.Map;
34+
import java.util.concurrent.ConcurrentHashMap;
35+
import java.util.function.Predicate;
3036

3137
/**
3238
* A collection class for handling metrics in {@link FlinkSourceReader} of Fluss.
@@ -64,9 +70,14 @@ public class FlinkSourceReaderMetrics {
6470
// Map for tracking current consuming offsets
6571
private final Map<TableBucket, Long> offsets = new HashMap<>();
6672

73+
// Map for tracking records lag of tableBucket
74+
@Nullable private Map<TableBucket, Metric> recordsLagMetrics;
75+
6776
// For currentFetchEventTimeLag metric
6877
private volatile long currentFetchEventTimeLag = UNINITIALIZED;
6978

79+
public static final String RECORDS_LAG = ".records-lag";
80+
7081
public FlinkSourceReaderMetrics(SourceReaderMetricGroup sourceReaderMetricGroup) {
7182
this.sourceReaderMetricGroup = sourceReaderMetricGroup;
7283
this.flussSourceReaderMetricGroup =
@@ -125,4 +136,51 @@ private void checkTableBucketTracked(TableBucket tableBucket) {
125136
public SourceReaderMetricGroup getSourceReaderMetricGroup() {
126137
return sourceReaderMetricGroup;
127138
}
139+
140+
public void maybeAddRecordsLagMetric(Map<String, ? extends Metric> metrics, TableBucket tb) {
141+
// Lazily register pendingRecords
142+
if (recordsLagMetrics == null) {
143+
this.recordsLagMetrics = new ConcurrentHashMap<>();
144+
this.sourceReaderMetricGroup.setPendingRecordsGauge(
145+
() -> {
146+
long pendingRecordsTotal = 0;
147+
for (Metric recordsLagMetric : this.recordsLagMetrics.values()) {
148+
pendingRecordsTotal +=
149+
Long.parseLong(
150+
((Gauge<?>) recordsLagMetric).getValue().toString());
151+
}
152+
return pendingRecordsTotal;
153+
});
154+
}
155+
recordsLagMetrics.computeIfAbsent(tb, (ignored) -> getRecordsLagMetric(metrics, tb));
156+
}
157+
158+
private @Nullable Metric getRecordsLagMetric(
159+
Map<String, ? extends Metric> metrics, TableBucket tb) {
160+
try {
161+
int bucket = tb.getBucket();
162+
Predicate<Map.Entry<String, ? extends Metric>> filter =
163+
entry -> {
164+
final String metricName = entry.getKey();
165+
return metricName.equals(bucket + RECORDS_LAG);
166+
};
167+
return metrics.entrySet().stream()
168+
.filter(filter)
169+
.map(Map.Entry::getValue)
170+
.findFirst()
171+
.orElseThrow(
172+
() ->
173+
new IllegalStateException(
174+
"Cannot find fluss metric matching current filter."));
175+
} catch (IllegalStateException e) {
176+
LOG.warn(
177+
String.format(
178+
"Error when getting fluss log scanner metric \"%s\" "
179+
+ "for bucket \"%s\". "
180+
+ "Metric \"%s\" may not be reported correctly. ",
181+
RECORDS_LAG, tb, MetricNames.PENDING_RECORDS),
182+
e);
183+
return null;
184+
}
185+
}
128186
}

‎fluss-connectors/fluss-connector-flink/src/main/java/com/alibaba/fluss/connector/flink/source/reader/FlinkSourceSplitReader.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -364,6 +364,7 @@ private FlinkRecordsWithSplitIds forLogRecords(ScanRecords scanRecords) {
364364
}
365365
}
366366
splitRecords.put(splitId, toRecordAndPos(bucketScanRecords.iterator()));
367+
flinkSourceReaderMetrics.maybeAddRecordsLagMetric(logScanner.metrics(), scanBucket);
367368
}
368369
Iterator<TableBucket> buckets = tableScanBuckets.iterator();
369370
Iterator<String> splitIterator =

‎fluss-connectors/fluss-connector-flink/src/test/java/com/alibaba/fluss/connector/flink/source/reader/FlinkSourceSplitReaderTest.java‎

Lines changed: 83 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,7 @@
4343
import org.apache.flink.connector.base.source.reader.splitreader.SplitsAddition;
4444
import org.apache.flink.connector.base.source.reader.splitreader.SplitsChange;
4545
import org.apache.flink.metrics.testutils.MetricListener;
46+
import org.apache.flink.runtime.metrics.MetricNames;
4647
import org.apache.flink.runtime.metrics.groups.InternalSourceReaderMetricGroup;
4748
import org.apache.flink.table.api.ValidationException;
4849
import org.junit.jupiter.api.Test;
@@ -207,6 +208,80 @@ void testHandleLogSplitChangesAndFetch() throws Exception {
207208
}
208209
}
209210

211+
@Test
212+
void testPendingRecords() throws Exception {
213+
final Schema schema =
214+
Schema.newBuilder()
215+
.column("id", DataTypes.INT())
216+
.column("name", DataTypes.STRING())
217+
.build();
218+
final TableDescriptor tableDescriptor =
219+
TableDescriptor.builder().schema(schema).distributedBy(1).build();
220+
TablePath tablePath1 = TablePath.of(DEFAULT_DB, "test-only-log-table");
221+
long tableId = createTable(tablePath1, tableDescriptor);
222+
MetricListener metricListener = new MetricListener();
223+
FlinkSourceReaderMetrics flinkSourceReaderMetrics =
224+
new FlinkSourceReaderMetrics(
225+
InternalSourceReaderMetricGroup.mock(metricListener.getMetricGroup()));
226+
227+
try (FlinkSourceSplitReader splitReader =
228+
createSplitReader(tablePath1, schema.toRowType(), flinkSourceReaderMetrics)) {
229+
230+
// no any records
231+
List<SourceSplitBase> logSplits = new ArrayList<>();
232+
Map<String, List<RecordAndPos>> expectedRecords = new HashMap<>();
233+
assignSplitsAndFetchUntilRetrieveRecords(
234+
splitReader, logSplits, expectedRecords, schema.toRowType());
235+
236+
assertThat(metricListener.getGauge(MetricNames.PENDING_RECORDS)).isNotPresent();
237+
238+
int rowCnt = 600;
239+
// now, write some records into the table
240+
List<InternalRow> internalRows = appendRows(tablePath1, rowCnt);
241+
List<RecordAndPos> expected = new ArrayList<>(internalRows.size());
242+
for (int i = 0; i < internalRows.size(); i++) {
243+
expected.add(
244+
new RecordAndPos(
245+
new ScanRecord(i, i, RowKind.APPEND_ONLY, internalRows.get(i))));
246+
}
247+
248+
TableBucket tableBucket = new TableBucket(tableId, 0);
249+
String splitId = toLogSplitId(tableBucket);
250+
expectedRecords.put(splitId, expected);
251+
252+
logSplits.add(new LogSplit(tableBucket, null, 0L));
253+
assignSplits(splitReader, logSplits);
254+
Set<String> finishedSplits = new HashSet<>();
255+
256+
int cnt = 0;
257+
while (finishedSplits.size() < logSplits.size()) {
258+
259+
RecordsWithSplitIds<RecordAndPos> recordsBySplitIds = splitReader.fetch();
260+
splitId = recordsBySplitIds.nextSplit();
261+
RecordAndPos record;
262+
if (splitId != null) {
263+
while ((record = recordsBySplitIds.nextRecordFromSplit()) != null) {
264+
cnt++;
265+
}
266+
// because lazy update this metric,client fetch all record in one batch
267+
assertThat(
268+
metricListener
269+
.getGauge(MetricNames.PENDING_RECORDS)
270+
.get()
271+
.getValue())
272+
.isEqualTo(600L);
273+
if (cnt >= rowCnt) {
274+
finishedSplits.add(splitId);
275+
}
276+
}
277+
}
278+
RecordsWithSplitIds<RecordAndPos> recordsBySplitIds = splitReader.fetch();
279+
// client update metric
280+
assertThat(metricListener.getGauge(MetricNames.PENDING_RECORDS).get().getValue())
281+
.isEqualTo(0L);
282+
}
283+
}
284+
210285
@Test
211286
void testHandleMixSnapshotLogSplitChangesAndFetch() throws Exception {
212287
TablePath tablePath = TablePath.of(DEFAULT_DB, "test-mix-snapshot-log-table");
@@ -343,6 +418,14 @@ private FlinkSourceSplitReader createSplitReader(TablePath tablePath, RowType ro
343418
clientConf, tablePath, rowType, null, createMockSourceReaderMetrics());
344419
}
345420

421+
private FlinkSourceSplitReader createSplitReader(
422+
TablePath tablePath,
423+
RowType rowType,
424+
FlinkSourceReaderMetrics flinkSourceReaderMetrics) {
425+
return new FlinkSourceSplitReader(
426+
clientConf, tablePath, rowType, null, flinkSourceReaderMetrics);
427+
}
428+
346429
private FlinkSourceReaderMetrics createMockSourceReaderMetrics() {
347430
MetricListener metricListener = new MetricListener();
348431
return new FlinkSourceReaderMetrics(

‎fluss-server/src/main/java/com/alibaba/fluss/server/metrics/group/BucketMetricGroup.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,7 @@ public class BucketMetricGroup extends AbstractMetricGroup {
2929

3030
private final int bucket;
3131

32-
public BucketMetricGroup(MetricRegistry registry, int bucket, PhysicalTableMetricGroup parent) {
32+
public BucketMetricGroup(MetricRegistry registry, int bucket, AbstractMetricGroup parent) {
3333
super(registry, makeScope(parent, String.valueOf(bucket)), parent);
3434
this.bucket = bucket;
3535
}

0 commit comments

Comments
 (0)