Skip to content

Commit d4840eb

Browse files
baisui1981baisui1981
committed
restore mongo split reader ability datavane/tis#456 
1 parent 41da77b commit d4840eb

6 files changed

Lines changed: 155 additions & 60 deletions

File tree

‎tis-incr/tis-flink-cdc-kingbase-plugin/src/test/java/com/qlangtech/plugins/incr/flink/cdc/kingbase/source/KingBaseConnectionPoolFactoryTest.java‎

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -19,9 +19,13 @@
1919
package com.qlangtech.plugins.incr.flink.cdc.kingbase.source;
2020

2121
import com.google.common.collect.Lists;
22+
import com.google.common.collect.Maps;
2223
import com.zaxxer.hikari.HikariDataSource;
24+
import io.debezium.config.Configuration;
2325
import org.apache.commons.collections.CollectionUtils;
26+
import org.apache.flink.cdc.connectors.base.options.StartupOptions;
2427
import org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceConfig;
28+
import org.jetbrains.annotations.Nullable;
2529
import org.junit.Assert;
2630
import org.junit.Test;
2731

@@ -30,6 +34,7 @@
3034
import java.sql.Statement;
3135
import java.time.Duration;
3236
import java.util.List;
37+
import java.util.Properties;
3338

3439
/**
3540
* @author: 百岁（baisui@qlangtech.com）
@@ -64,8 +69,13 @@ public void testCreatePooledDataSource() throws Exception {
6469
*/
6570

6671
private static class StubPostgresSourceConfig extends PostgresSourceConfig {
67-
public StubPostgresSourceConfig() {
68-
super(0, null, Lists.newArrayList("test"), Lists.newArrayList("public"), null, 0, 0, 0d, 0d, true, true, null, null, null, "192.168.28.201", 4321, "kingbase", "123456", 0, "Asia/Shanghai", Duration.ofSeconds(10), 1, 1, null, true, true);
72+
// public StubPostgresSourceConfig() {
73+
// super(0, null, Lists.newArrayList("test"), Lists.newArrayList("public"), null, 0, 0, 0d, 0d, true, true, null, null, null, "192.168.28.201", 4321, "kingbase", "123456", 0, "Asia/Shanghai", Duration.ofSeconds(10), 1, 1, null, true, true);
74+
// }
75+
76+
public StubPostgresSourceConfig( //int subtaskId, StartupOptions startupOptions, List<String> databaseList, List<String> schemaList, List<String> tableList, int splitSize, int splitMetaGroupSize, double distributionFactorUpper, double distributionFactorLower, boolean includeSchemaChanges, boolean closeIdleReaders, Properties dbzProperties, Configuration dbzConfiguration, String driverClassName, String hostname, int port, String username, String password, int fetchSize, String serverTimeZone, Duration connectTimeout, int connectMaxRetries, int connectionPoolSize, @Nullable String chunkKeyColumn, boolean skipSnapshotBackfill, boolean isScanNewlyAddedTableEnabled, int lsnCommitCheckpointsDelay, boolean assignUnboundedChunkFirst
77+
) {
78+
super(0, null, Lists.newArrayList("test"), Lists.newArrayList("public"), null, 0, 0, 0d, 0d, true, true, new Properties(), Configuration.from(Maps.newHashMap()), null, "192.168.28.201", 4321, "kingbase", "123456", 0, "Asia/Shanghai", Duration.ofSeconds(10), 1, 1, null, true, false, 1, true);
6979
}
7080
}
7181
}

‎tis-incr/tis-flink-cdc-mongdb-plugin/src/test/java/com/qlangtech/plugins/incr/flink/cdc/mongdb/TestTISFlinkCDCMongoDBSourceFunction.java‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -59,6 +59,8 @@
5959
import com.qlangtech.tis.plugin.ds.mangodb.MangoDBDataSourceFactory;
6060
import com.qlangtech.tis.plugin.ds.mangodb.MangoDBDataSourceFactory.MongoJDBCConnection;
6161
import com.qlangtech.tis.plugin.incr.TISSinkFactory;
62+
import com.qlangtech.tis.plugin.timezone.DefaultTISTimeZone;
63+
import com.qlangtech.tis.plugin.timezone.TISTimeZone;
6264
import com.qlangtech.tis.sql.parser.tuple.creator.EntityName;
6365
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
6466
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -169,7 +171,9 @@ public MangoDBDataSourceFactory getDataSourceFactory() {
169171
}
170172
};
171173
mongoReader.selectedTabs = rdbmsReader.selectedTabs;
172-
mongoReader.timeZone = MQListenerFactory.dftZoneId();
174+
DefaultTISTimeZone timeZone = new DefaultTISTimeZone();
175+
timeZone.timeZone = TISTimeZone.dftZoneId();
176+
mongoReader.timeZone = timeZone;
173177
return mongoReader;
174178
// DataSourceFactory dataSourceFactory = createDataSourceFactory(dataxName, this.splitTabSuffix.isPresent());
175179
// BasicDataXRdbmsReader dataxReader = new BasicDataXRdbmsReader() {

‎tis-incr/tis-flink-extends/src/main/java/com/qlangtech/plugins/incr/flink/metrics/TISPBReporter.java‎

Lines changed: 87 additions & 45 deletions
Original file line numberDiff line numberDiff line change
@@ -21,12 +21,13 @@
2121
import com.google.common.collect.Maps;
2222
import com.google.common.collect.Lists;
2323
import com.qlangtech.tis.cloud.ITISCoordinator;
24-
import com.qlangtech.tis.realtime.SourceProcessFunction;
2524
import com.qlangtech.tis.realtime.transfer.IIncreaseCounter;
25+
import com.qlangtech.tis.realtime.transfer.ListenerStatusKeeper.LimitRateTypeAndRatePerSecNums;
2626
import com.qlangtech.tis.realtime.transfer.TableSingleDataIndexStatus;
27-
import com.qlangtech.tis.realtime.yarn.rpc.MasterJob;
27+
import com.qlangtech.tis.realtime.yarn.rpc.IncrRateControllerCfgDTO.RateControllerType;
2828
import com.qlangtech.tis.realtime.yarn.rpc.PipelineFlinkTaskId;
2929
import com.qlangtech.tis.realtime.yarn.rpc.UpdateCounterMap;
30+
3031
import com.tis.hadoop.rpc.RpcServiceReference;
3132
import com.tis.hadoop.rpc.StatusRpcClientFactory;
3233

@@ -35,9 +36,11 @@
3536
import org.apache.commons.lang3.tuple.Pair;
3637
import org.apache.flink.metrics.CharacterFilter;
3738
import org.apache.flink.metrics.Counter;
39+
import org.apache.flink.metrics.Gauge;
3840
import org.apache.flink.metrics.Metric;
3941
import org.apache.flink.metrics.MetricConfig;
4042
import org.apache.flink.metrics.MetricGroup;
43+
import org.apache.flink.metrics.MetricType;
4144
import org.apache.flink.metrics.reporter.AbstractReporter;
4245
import org.apache.flink.metrics.reporter.Scheduled;
4346
import org.apache.flink.runtime.metrics.scope.ScopeFormat;
@@ -48,14 +51,22 @@
4851
import java.util.Optional;
4952
import java.util.concurrent.ConcurrentMap;
5053

54+
import static com.qlangtech.tis.realtime.transfer.IIncreaseCounter.METRIC_LIMIT_RATE_CONTROLLER_TYPE;
55+
import static com.qlangtech.tis.realtime.transfer.IIncreaseCounter.METRIC_LIMIT_RATE_PER_SECOND_NUMS;
56+
5157
/**
5258
* https://github.com/datavane/tis/issues/397
5359
*
5460
* @author: 百岁（baisui@qlangtech.com）
5561
* @create: 2025-05-14 10:24
5662
**/
5763
public class TISPBReporter extends AbstractReporter implements Scheduled {
58-
private final ConcurrentMap<String /**metric identity*/, Pair<String /**metricName*/, MetricGroup>> metricIdentifierMapper = Maps.newConcurrentMap();
64+
65+
private final ConcurrentMap<String /**metric identity*/, Pair<String /**metricName*/, MetricGroup>>
66+
counterMetricIdentifierMapper = Maps.newConcurrentMap();
67+
68+
private final ConcurrentMap<String /**metric identity*/, Pair<String /**metricName*/, MetricGroup>>
69+
gaugeMetricIdentifierMapper = Maps.newConcurrentMap();
5970

6071
//private static final ConcurrentMap<String /**taskId*/, MasterJob> receivedTaskControllerMessage = Maps.newConcurrentMap();
6172

@@ -112,16 +123,24 @@ public void notifyOfAddedMetric(Metric metric, String metricName, MetricGroup gr
112123
if (scope.length > 1 && TaskExecutor.TASK_MANAGER_NAME.equals(scope[1])) {
113124
if (IIncreaseCounter.COLLECTABLE_TABLE_COUNT_METRIC.contains(metricName)) {
114125
// System.out.println(metricName);
115-
metricIdentifierMapper.put(name, Pair.of(metricName, group));
126+
counterMetricIdentifierMapper.put(name, Pair.of(metricName, group));
127+
} else if (IIncreaseCounter.COLLECTABLE_METRIC_LIMIT_GAUGE.contains(metricName)) {
128+
gaugeMetricIdentifierMapper.put(name, Pair.of(metricName, group));
116129
}
130+
117131
}
118132
}
119133

120134
@Override
121135
public void notifyOfRemovedMetric(Metric metric, String metricName, MetricGroup group) {
122136
super.notifyOfRemovedMetric(metric, metricName, group);
123137
String name = group.getMetricIdentifier(metricName, this);
124-
metricIdentifierMapper.remove(name);
138+
if (metric.getMetricType() == MetricType.COUNTER) {
139+
counterMetricIdentifierMapper.remove(name);
140+
} else if (metric.getMetricType() == MetricType.GAUGE) {
141+
gaugeMetricIdentifierMapper.remove(name);
142+
}
143+
125144
}
126145

127146
@Override
@@ -131,46 +150,26 @@ public void close() {
131150

132151
@Override
133152
public void report() {
134-
// UpdateCounterMap upateCounter = null;
135-
//
136-
// UpdateCounterMap updateCounterMap = new UpdateCounterMap();
137-
// updateCounterMap.setGcCounter(BasicONSListener.getGarbageCollectionCount());
138-
// updateCounterMap.setFrom(hostName);
139-
// long currentTimeInSec = ConsumeDataKeeper.getCurrentTimeInSec();
140-
// updateCounterMap.setUpdateTime(currentTimeInSec);
141-
// 汇总一个节点中所有索引的增量信息
142-
// for (IOnsListenerStatus l : incrChannels) {
143-
// TableSingleDataIndexStatus tableUpdateCounter = new TableSingleDataIndexStatus();
144-
// tableUpdateCounter.setBufferQueueRemainingCapacity(l.getBufferQueueRemainingCapacity());
145-
// tableUpdateCounter.setBufferQueueUsedSize(l.getBufferQueueUsedSize());
146-
// tableUpdateCounter.setConsumeErrorCount((int) l.getConsumeErrorCount());
147-
// tableUpdateCounter.setIgnoreRowsCount((int) l.getIgnoreRowsCount());
148-
// tableUpdateCounter.setUUID(this.indexUUID.get(l.getCollectionName()));
149-
// tableUpdateCounter.setTis30sAvgRT(((BasicONSListener) l).getTis30sAvgRT());
150-
// // 汇总一个索引中所有focus table的增量信息
151-
// for (Map.Entry<String, IIncreaseCounter> entry : l.getUpdateStatic()) {
152-
// // IncrCounter tableIncrCounter = new
153-
// // IncrCounter((int)entry.getValue().getIncreasePastLast());
154-
// // tableIncrCounter.setAccumulationCount(entry.getValue().getAccumulation());
155-
// // tableUpdateCounter.put(entry.getKey(), tableIncrCounter);
156-
// // 只记录一个消费总量和当前时间
157-
// tableUpdateCounter.put(entry.getKey(), entry.getValue().getAccumulation());
158-
// }
159-
// tableUpdateCounter.put(TABLE_CONSUME_COUNT, ((BasicONSListener) l).getTableConsumeCount());
160-
// updateCounterMap.addTableCounter(l.getCollectionName(), tableUpdateCounter);
161-
// }
162153
Pair<String /**metricName*/, MetricGroup> metricGroup = null;
163154
// MasterJob masterJob = this.rpcService.reportStatus(upateCounter);
164155
List<UseableMetricForTIS> metrics = Lists.newArrayList();
156+
String metricID = null;
165157
for (Map.Entry<Counter, String> entry : counters.entrySet()) {
166158
Counter counter = entry.getKey();
167-
String metricID = entry.getValue();
159+
metricID = entry.getValue();
168160
// System.out.println(metricID + ": " + counter.getCount());
169-
metricGroup = metricIdentifierMapper.get(metricID);
161+
metricGroup = counterMetricIdentifierMapper.get(metricID);
170162
if (metricGroup != null) {
171-
// System.out.println(metricGroup);
163+
metrics.add(new UseableCounterMetricForTIS(counter, /**metricName*/metricGroup.getKey(), metricGroup.getRight()));
164+
}
165+
}
172166

173-
metrics.add(new UseableMetricForTIS(counter, /**metricName*/metricGroup.getKey(), metricGroup.getRight()));
167+
for (Map.Entry<Gauge<?>, String> entry : this.gauges.entrySet()) {
168+
Gauge<?> gauge = entry.getKey();
169+
metricID = entry.getValue();
170+
metricGroup = gaugeMetricIdentifierMapper.get(metricID);
171+
if (metricGroup != null) {
172+
metrics.add(new UseableGaugeMetricForTIS(gauge, /**metricName*/metricGroup.getKey(), metricGroup.getRight()));
174173
}
175174
}
176175

@@ -180,7 +179,7 @@ public void report() {
180179
private void sendMetric2TISAssemble(List<UseableMetricForTIS> metrics) {
181180
// 汇总一个索引中所有focus table的增量信息
182181
Map<PipelineFlinkTaskId, TableSingleDataIndexStatus> tabCounterMapper = Maps.newHashMap();
183-
182+
UpdateCounterMap pipelineUpdateCounterMap = new UpdateCounterMap();
184183
PipelineFlinkTaskId pipelineName = null;
185184
TableSingleDataIndexStatus singleDataIndexStatus = null;
186185
String host = null;
@@ -195,14 +194,18 @@ private void sendMetric2TISAssemble(List<UseableMetricForTIS> metrics) {
195194
singleDataIndexStatus.setUUID(metric.getFlinkTaskId());
196195
tabCounterMapper.put(pipelineName, singleDataIndexStatus);
197196
}
198-
singleDataIndexStatus.put(metric.metricName, metric.counter.getCount());
197+
metric.putCounterMetric(singleDataIndexStatus);
198+
// metric.putGaugeMetric(pipelineUpdateCounterMap);
199199
}
200200

201201
if (MapUtils.isNotEmpty(tabCounterMapper)) {
202-
UpdateCounterMap pipelineUpdateCounterMap = new UpdateCounterMap();
202+
203203
if (StringUtils.isEmpty(host)) {
204204
throw new IllegalStateException("host can not be empty");
205205
}
206+
// pipelineUpdateCounterMap.setGcCounter();
207+
208+
206209
pipelineUpdateCounterMap.setFrom(host);
207210
tabCounterMapper.forEach((tisPipeline, tabCounter) -> {
208211

@@ -215,9 +218,9 @@ private void sendMetric2TISAssemble(List<UseableMetricForTIS> metrics) {
215218
}
216219
}
217220

218-
private static class UseableMetricForTIS {
219-
private Counter counter;
220-
private String metricName;
221+
private static abstract class UseableMetricForTIS<METRIC_TYPE> {
222+
protected METRIC_TYPE metric;
223+
protected String metricName;
221224
private MetricGroup metricGroup;
222225

223226
String getPipelineName() {
@@ -232,10 +235,49 @@ String getFlinkTaskId() {
232235
return metricGroup.getAllVariables().get(ScopeFormat.SCOPE_TASK_SUBTASK_INDEX);
233236
}
234237

235-
public UseableMetricForTIS(Counter counter, String metricName, MetricGroup metricGroup) {
236-
this.counter = counter;
238+
public UseableMetricForTIS(METRIC_TYPE counter, String metricName, MetricGroup metricGroup) {
239+
this.metric = counter;
237240
this.metricName = metricName;
238241
this.metricGroup = metricGroup;
239242
}
243+
244+
public abstract void putCounterMetric(TableSingleDataIndexStatus singleDataIndexStatus);
245+
246+
// public abstract void putGaugeMetric(UpdateCounterMap pipelineUpdateCounterMap);
247+
}
248+
249+
private static class UseableCounterMetricForTIS extends UseableMetricForTIS<Counter> {
250+
public UseableCounterMetricForTIS(Counter counter, String metricName, MetricGroup metricGroup) {
251+
super(counter, metricName, metricGroup);
252+
}
253+
254+
@Override
255+
public void putCounterMetric(TableSingleDataIndexStatus singleDataIndexStatus) {
256+
singleDataIndexStatus.put(this.metricName, this.metric.getCount());
257+
}
258+
}
259+
260+
private static class UseableGaugeMetricForTIS extends UseableMetricForTIS<Gauge<?>> {
261+
public UseableGaugeMetricForTIS(Gauge<?> gauge, String metricName, MetricGroup metricGroup) {
262+
super(gauge, metricName, metricGroup);
263+
}
264+
265+
@Override
266+
public void putCounterMetric(TableSingleDataIndexStatus singleDataIndexStatus) {
267+
LimitRateTypeAndRatePerSecNums origin = singleDataIndexStatus.getIncrRateLimitConfig();
268+
LimitRateTypeAndRatePerSecNums config = null;
269+
270+
if (METRIC_LIMIT_RATE_CONTROLLER_TYPE.equals(this.metricName)) {
271+
config = new LimitRateTypeAndRatePerSecNums(RateControllerType.parse((short) this.metric.getValue()), origin.getPerSecRateNums());
272+
} else if (METRIC_LIMIT_RATE_PER_SECOND_NUMS.equals(this.metricName)) {
273+
config = new LimitRateTypeAndRatePerSecNums(origin.getControllerType().orElse(null), (int) this.metric.getValue());
274+
} else {
275+
throw new IllegalStateException("illegal metricName:" + this.metricName);
276+
}
277+
278+
singleDataIndexStatus.setIncrRateLimitConfig(config);
279+
}
280+
281+
240282
}
241283
}

‎tis-incr/tis-flink-extends/src/main/java/com/qlangtech/tis/realtime/ReaderSource.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -106,7 +106,7 @@ private static class EmptyGeneratorFunction implements GeneratorFunction<Long, I
106106
public IncrRateControllerCfgDTO map(Long aLong) throws Exception {
107107
IncrRateControllerCfgDTO mockRateControllerCfg = new IncrRateControllerCfgDTO();
108108
mockRateControllerCfg.setControllerType(RateControllerType.SkipProcess);
109-
mockRateControllerCfg.setPause(false);
109+
//mockRateControllerCfg.setPause(false);
110110
mockRateControllerCfg.setLastModified(System.currentTimeMillis());
111111
return mockRateControllerCfg;
112112
}

0 commit comments

Comments
 (0)