chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

Canal metrics snapshot version.

Browse Source
This commit is contained in:
Chuanyi Li
2018-08-07 22:52:30 +08:00
parent fec9803aae
commit 179afb0340
14 changed files with 85 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
deployer/pom.xml
View File

@@ -18,12 +18,12 @@
</dependency> </dependency>
<!-- 这里指定runtime的metrics provider--> <!-- 这里指定runtime的metrics provider-->
<!--<dependency>--> <dependency>
<!--<groupId>com.alibaba.otter</groupId>--> <groupId>com.alibaba.otter</groupId>
<!--<artifactId>canal.prometheus</artifactId>--> <artifactId>canal.prometheus</artifactId>
<!--<version>${project.version}</version>--> <version>${project.version}</version>
<!--<scope>runtime</scope>--> <scope>runtime</scope>
<!--</dependency>--> </dependency>
</dependencies> </dependencies>
<build> <build>

15
deployer/src/main/bin/metrics_env.sh
View File

@@ -1,15 +0,0 @@
#!/bin/bash
# Additional line arg for current prometheus solution
case "`uname`" in
Linux)
bin_abs_path=$(readlink -f $(dirname $0))
;;
*)
bin_abs_path=`cd $(dirname $0); pwd`
;;
esac
base=${bin_abs_path}/..
if [ $(ls $base/lib/aspectjweaver*.jar | wc -l) -eq 1 ]; then
WEAVER=$(ls $base/lib/aspectjweaver*.jar)
METRICS_OPTS=" -javaagent:"${WEAVER}" "
fi

7
deployer/src/main/bin/startup.sh
View File

@@ -94,12 +94,7 @@ then
echo LOG CONFIGURATION : $logback_configurationFile echo LOG CONFIGURATION : $logback_configurationFile
echo canal conf : $canal_conf echo canal conf : $canal_conf
echo CLASSPATH :$CLASSPATH echo CLASSPATH :$CLASSPATH
# metrics support options $JAVA $JAVA_OPTS $JAVA_DEBUG_OPT $CANAL_OPTS -classpath .:$CLASSPATH com.alibaba.otter.canal.deployer.CanalLauncher 1>>$base/logs/canal/canal.log 2>&1 &
# if [ -x $base/bin/metrics_env.sh ]; then
# . $base/bin/metrics_env.sh
# echo METRICS_OPTS $METRICS_OPTS
# fi
$JAVA $JAVA_OPTS $METRICS_OPTS $JAVA_DEBUG_OPT $CANAL_OPTS -classpath .:$CLASSPATH com.alibaba.otter.canal.deployer.CanalLauncher 1>>$base/logs/canal/canal.log 2>&1 &
echo $! > $base/bin/canal.pid echo $! > $base/bin/canal.pid
echo "cd to $current_path for continue" echo "cd to $current_path for continue"

1
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java
View File

@@ -52,6 +52,7 @@ public class MysqlConnection implements ErosaConnection {
private AuthenticationInfo authInfo; private AuthenticationInfo authInfo;
protected int connTimeout = 5 * 1000; // 5秒 protected int connTimeout = 5 * 1000; // 5秒
protected int soTimeout = 60 * 60 * 1000; // 1小时 protected int soTimeout = 60 * 60 * 1000; // 1小时
// dump binlog bytes, 暂不包括meta与TSDB
private AtomicLong receivedBinlogBytes; private AtomicLong receivedBinlogBytes;
public MysqlConnection(){ public MysqlConnection(){

5
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java
View File

@@ -187,6 +187,11 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
//LockSupport.parkNanos(1L); //LockSupport.parkNanos(1L);
applyWait(++fullTimes); applyWait(++fullTimes);
interupted = Thread.interrupted(); interupted = Thread.interrupted();
if (fullTimes % 1000 == 0) {
long nextStart = System.nanoTime();
eventsPublishBlockingTime.addAndGet(nextStart - blockingStart);
blockingStart = nextStart;
}
} }
} while (!interupted && isStart()); } while (!interupted && isStart());
return isStart(); return isStart();

2
prometheus/src/main/java/com/alibaba/otter/canal/prometheus/CanalInstanceExports.java
View File

@@ -77,7 +77,7 @@ public class CanalInstanceExports {
private InstanceRegistry requiredInstanceRegistry(Collector collector) { private InstanceRegistry requiredInstanceRegistry(Collector collector) {
if (!(collector instanceof InstanceRegistry)) { if (!(collector instanceof InstanceRegistry)) {
throw new IllegalArgumentException("Canal collector need to implement InstanceRegistry."); throw new IllegalArgumentException("Canal prometheus collector need to implement InstanceRegistry.");
} }
return (InstanceRegistry) collector; return (InstanceRegistry) collector;
} }

3
prometheus/src/main/java/com/alibaba/otter/canal/prometheus/PrometheusService.java
View File

@@ -23,10 +23,11 @@ public class PrometheusService implements CanalMetricsService {
private final CanalInstanceExports instanceExports; private final CanalInstanceExports instanceExports;
private volatile boolean running = false; private volatile boolean running = false;
private HTTPServer server; private HTTPServer server;
private ClientInstanceProfiler clientProfiler = new PrometheusClientInstanceProfiler(); private final ClientInstanceProfiler clientProfiler;
private PrometheusService() { private PrometheusService() {
this.instanceExports = CanalInstanceExports.instance(); this.instanceExports = CanalInstanceExports.instance();
this.clientProfiler = PrometheusClientInstanceProfiler.instance();
} }
private static class SingletonHolder { private static class SingletonHolder {

8
prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/EntryCollector.java
View File

@@ -29,7 +29,7 @@ public class EntryCollector extends Collector implements InstanceRegistry {
private static final Logger logger = LoggerFactory.getLogger(SinkCollector.class); private static final Logger logger = LoggerFactory.getLogger(SinkCollector.class);
private static final String DELAY = "canal_instance_traffic_delay"; private static final String DELAY = "canal_instance_traffic_delay";
private static final String TRANSACTION = "canal_instance_transactions"; private static final String TRANSACTION = "canal_instance_transactions";
private static final String DELAY_HELP = "Traffic delay of canal instance"; private static final String DELAY_HELP = "Traffic delay of canal instance in milliseconds";
private static final String TRANSACTION_HELP = "Transactions counter of canal instance"; private static final String TRANSACTION_HELP = "Transactions counter of canal instance";
private final ConcurrentMap<String, EntryMetricsHolder> instances = new ConcurrentHashMap<String, EntryMetricsHolder>(); private final ConcurrentMap<String, EntryMetricsHolder> instances = new ConcurrentHashMap<String, EntryMetricsHolder>();
@@ -51,7 +51,11 @@ public class EntryCollector extends Collector implements InstanceRegistry {
CounterMetricFamily transactions = new CounterMetricFamily(TRANSACTION, CounterMetricFamily transactions = new CounterMetricFamily(TRANSACTION,
TRANSACTION_HELP, DEST_LABELS_LIST); TRANSACTION_HELP, DEST_LABELS_LIST);
for (EntryMetricsHolder emh : instances.values()) { for (EntryMetricsHolder emh : instances.values()) {
delay.addMetric(emh.destLabelValues, emh.latestExecTime.doubleValue()); long now = System.currentTimeMillis();
long latest = emh.latestExecTime.get();
if (now > latest) {
delay.addMetric(emh.destLabelValues, (now - latest));
}
transactions.addMetric(emh.destLabelValues, emh.transactionCounter.doubleValue()); transactions.addMetric(emh.destLabelValues, emh.transactionCounter.doubleValue());
} }
mfs.add(delay); mfs.add(delay);

29
prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/MetaCollector.java
View File

@@ -11,22 +11,23 @@ import io.prometheus.client.GaugeMetricFamily;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import java.util.ArrayList; import java.util.*;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ConcurrentMap;
import static com.alibaba.otter.canal.prometheus.CanalInstanceExports.DEST_LABELS_LIST;
/** /**
* @author Chuanyi Li * @author Chuanyi Li
*/ */
public class MetaCollector extends Collector implements InstanceRegistry { public class MetaCollector extends Collector implements InstanceRegistry {
private static final List<String> InfoLabel = Arrays.asList("destination", "mode"); private static final List<String> INFO_LABELS_LIST = Arrays.asList("destination", "mode");
private static final Logger logger = LoggerFactory.getLogger(MetaCollector.class); private static final Logger logger = LoggerFactory.getLogger(MetaCollector.class);
private static final String SUBSCRIPTION = "canal_instance"; private static final String INSTANCE = "canal_instance";
private static final String SUBSCRIPTION_HELP = "Canal instance"; private static final String INSTANCE_HELP = "Canal instance";
private static final String SUBSCRIPTION = "canal_instance_subscriptions";
private static final String SUBSCRIPTION_HELP = "Canal instance subscriptions";
private final ConcurrentMap<String, MetaMetricsHolder> instances = new ConcurrentHashMap<String, MetaMetricsHolder>(); private final ConcurrentMap<String, MetaMetricsHolder> instances = new ConcurrentHashMap<String, MetaMetricsHolder>();
private MetaCollector() {} private MetaCollector() {}
@@ -42,18 +43,20 @@ public class MetaCollector extends Collector implements InstanceRegistry {
@Override @Override
public List<MetricFamilySamples> collect() { public List<MetricFamilySamples> collect() {
List<MetricFamilySamples> mfs = new ArrayList<MetricFamilySamples>(); List<MetricFamilySamples> mfs = new ArrayList<MetricFamilySamples>();
GaugeMetricFamily instanceInfo = new GaugeMetricFamily( GaugeMetricFamily instanceInfo = new GaugeMetricFamily(INSTANCE,
SUBSCRIPTION, INSTANCE_HELP, INFO_LABELS_LIST);
SUBSCRIPTION_HELP, GaugeMetricFamily subsInfo = new GaugeMetricFamily(SUBSCRIPTION,
InfoLabel); SUBSCRIPTION_HELP, DEST_LABELS_LIST);
for (Map.Entry<String, MetaMetricsHolder> nme : instances.entrySet()) { for (Map.Entry<String, MetaMetricsHolder> nme : instances.entrySet()) {
final String destination = nme.getKey(); final String destination = nme.getKey();
final MetaMetricsHolder nmh = nme.getValue(); final MetaMetricsHolder nmh = nme.getValue();
instanceInfo.addMetric(nmh.infoLabelValues, 1);
List<ClientIdentity> subs = nmh.metaManager.listAllSubscribeInfo(destination); List<ClientIdentity> subs = nmh.metaManager.listAllSubscribeInfo(destination);
int count = subs == null ? 0 : subs.size(); int count = subs == null ? 0 : subs.size();
instanceInfo.addMetric(nmh.infoLabelValues, count); subsInfo.addMetric(nmh.destLabelValues, count);
} }
mfs.add(instanceInfo); mfs.add(instanceInfo);
mfs.add(subsInfo);
return mfs; return mfs;
} }
@@ -63,6 +66,7 @@ public class MetaCollector extends Collector implements InstanceRegistry {
MetaMetricsHolder holder = new MetaMetricsHolder(); MetaMetricsHolder holder = new MetaMetricsHolder();
String mode = (instance instanceof CanalInstanceWithSpring) ? "spring" : "manager"; String mode = (instance instanceof CanalInstanceWithSpring) ? "spring" : "manager";
holder.infoLabelValues = Arrays.asList(destination, mode); holder.infoLabelValues = Arrays.asList(destination, mode);
holder.destLabelValues = Collections.singletonList(destination);
holder.metaManager = instance.getMetaManager(); holder.metaManager = instance.getMetaManager();
Preconditions.checkNotNull(holder.metaManager); Preconditions.checkNotNull(holder.metaManager);
MetaMetricsHolder old = instances.put(destination, holder); MetaMetricsHolder old = instances.put(destination, holder);
@@ -79,6 +83,7 @@ public class MetaCollector extends Collector implements InstanceRegistry {
private class MetaMetricsHolder { private class MetaMetricsHolder {
private List<String> infoLabelValues; private List<String> infoLabelValues;
private List<String> destLabelValues;
private CanalMetaManager metaManager; private CanalMetaManager metaManager;
} }

6
prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/ParserCollector.java
View File

@@ -21,6 +21,7 @@ import java.util.concurrent.atomic.AtomicLong;
import static com.alibaba.otter.canal.prometheus.CanalInstanceExports.DEST; import static com.alibaba.otter.canal.prometheus.CanalInstanceExports.DEST;
import static com.alibaba.otter.canal.prometheus.CanalInstanceExports.DEST_LABELS_LIST; import static com.alibaba.otter.canal.prometheus.CanalInstanceExports.DEST_LABELS_LIST;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
/** /**
* @author Chuanyi Li * @author Chuanyi Li
@@ -28,11 +29,12 @@ import static com.alibaba.otter.canal.prometheus.CanalInstanceExports.DEST_LABEL
public class ParserCollector extends Collector implements InstanceRegistry { public class ParserCollector extends Collector implements InstanceRegistry {
private static final Logger logger = LoggerFactory.getLogger(ParserCollector.class); private static final Logger logger = LoggerFactory.getLogger(ParserCollector.class);
private static final long NANO_PER_MILLI = 1000 * 1000L;
private static final String PUBLISH_BLOCKING = "canal_instance_publish_blocking_time"; private static final String PUBLISH_BLOCKING = "canal_instance_publish_blocking_time";
private static final String RECEIVED_BINLOG = "canal_instance_received_binlog_bytes"; private static final String RECEIVED_BINLOG = "canal_instance_received_binlog_bytes";
private static final String PARSER_MODE = "canal_instance_parser_mode"; private static final String PARSER_MODE = "canal_instance_parser_mode";
private static final String MODE_LABEL = "parallel"; private static final String MODE_LABEL = "parallel";
private static final String PUBLISH_BLOCKING_HELP = "Publish blocking time of dump thread"; private static final String PUBLISH_BLOCKING_HELP = "Publish blocking time of dump thread in milliseconds";
private static final String RECEIVED_BINLOG_HELP = "Received binlog bytes"; private static final String RECEIVED_BINLOG_HELP = "Received binlog bytes";
private static final String MODE_HELP = "Parser mode(parallel/serial) of instance"; private static final String MODE_HELP = "Parser mode(parallel/serial) of instance";
private final List<String> modeLabels = Arrays.asList(DEST, MODE_LABEL); private final List<String> modeLabels = Arrays.asList(DEST, MODE_LABEL);
@@ -60,7 +62,7 @@ public class ParserCollector extends Collector implements InstanceRegistry {
PUBLISH_BLOCKING_HELP, DEST_LABELS_LIST); PUBLISH_BLOCKING_HELP, DEST_LABELS_LIST);
for (ParserMetricsHolder emh : instances.values()) { for (ParserMetricsHolder emh : instances.values()) {
if (emh.isParallel) { if (emh.isParallel) {
blockingCounter.addMetric(emh.destLabelValues, emh.eventsPublishBlockingTime.doubleValue()); blockingCounter.addMetric(emh.destLabelValues, (emh.eventsPublishBlockingTime.doubleValue() / NANO_PER_MILLI));
hasParallel = true; hasParallel = true;
} }
modeGauge.addMetric(emh.modeLabelValues, 1); modeGauge.addMetric(emh.modeLabelValues, 1);

19
prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/PrometheusCanalEventDownStreamHandler.java
View File

@@ -1,5 +1,6 @@
package com.alibaba.otter.canal.prometheus.impl; package com.alibaba.otter.canal.prometheus.impl;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType; import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.sink.AbstractCanalEventDownStreamHandler; import com.alibaba.otter.canal.sink.AbstractCanalEventDownStreamHandler;
import com.alibaba.otter.canal.store.model.Event; import com.alibaba.otter.canal.store.model.Event;
@@ -24,18 +25,28 @@ public class PrometheusCanalEventDownStreamHandler extends AbstractCanalEventDow
for (Event e : events) { for (Event e : events) {
EntryType type = e.getEntryType(); EntryType type = e.getEntryType();
if (type == null) continue; if (type == null) continue;
switch (type) {
case TRANSACTIONBEGIN: {
long exec = e.getExecuteTime();
if (exec > 0) localExecTime = exec;
break;
}
case ROWDATA: {
long exec = e.getExecuteTime(); long exec = e.getExecuteTime();
if (exec > 0) localExecTime = exec; if (exec > 0) localExecTime = exec;
switch (type) {
case ROWDATA:
// TODO 当前proto无法直接获得荣威change的变更行数（需要parse），可考虑放到header里面 // TODO 当前proto无法直接获得荣威change的变更行数（需要parse），可考虑放到header里面
break; break;
case TRANSACTIONEND: }
case TRANSACTIONEND: {
long exec = e.getExecuteTime();
if (exec > 0) localExecTime = exec;
transactionCounter.incrementAndGet(); transactionCounter.incrementAndGet();
break; break;
}
case HEARTBEAT: case HEARTBEAT:
// 发现canal自己的heartbeat是带有execTime的
// TODO 确认一下不是canal自己产生的 // TODO 确认一下不是canal自己产生的
// EventType eventType = e.getEventType(); CanalEntry.EventType eventType = e.getEventType();
// TODO utilize MySQL master heartbeat packet to refresh delay if always no more events coming // TODO utilize MySQL master heartbeat packet to refresh delay if always no more events coming
// see: https://dev.mysql.com/worklog/task/?id=342 // see: https://dev.mysql.com/worklog/task/?id=342
// heartbeats are sent by the master only if there is no // heartbeats are sent by the master only if there is no

15
prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/PrometheusClientInstanceProfiler.java
View File

@@ -15,6 +15,7 @@ import static com.alibaba.otter.canal.prometheus.CanalInstanceExports.DEST_LABEL
*/ */
public class PrometheusClientInstanceProfiler implements ClientInstanceProfiler { public class PrometheusClientInstanceProfiler implements ClientInstanceProfiler {
private static final long NANO_PER_MILLI = 1000 * 1000L;
private static final String PACKET_TYPE = "canal_instance_client_packets"; private static final String PACKET_TYPE = "canal_instance_client_packets";
private static final String OUTBOUND_BYTES = "canal_instance_client_bytes"; private static final String OUTBOUND_BYTES = "canal_instance_client_bytes";
private static final String EMPTY_BATCHES = "canal_instance_client_empty_batches"; private static final String EMPTY_BATCHES = "canal_instance_client_empty_batches";
@@ -27,7 +28,15 @@ public class PrometheusClientInstanceProfiler implements ClientInstanceProfiler
private final Histogram responseLatency; private final Histogram responseLatency;
private volatile boolean running = false; private volatile boolean running = false;
public PrometheusClientInstanceProfiler() { private static class SingletonHolder {
private static final PrometheusClientInstanceProfiler SINGLETON = new PrometheusClientInstanceProfiler();
}
public static PrometheusClientInstanceProfiler instance() {
return SingletonHolder.SINGLETON;
}
private PrometheusClientInstanceProfiler() {
this.outboundCounter = Counter.build() this.outboundCounter = Counter.build()
.labelNames(DEST_LABELS) .labelNames(DEST_LABELS)
.name(OUTBOUND_BYTES) .name(OUTBOUND_BYTES)
@@ -53,7 +62,7 @@ public class PrometheusClientInstanceProfiler implements ClientInstanceProfiler
.name(LATENCY) .name(LATENCY)
.help("Client request latency.") .help("Client request latency.")
// buckets in milliseconds // buckets in milliseconds
.buckets(1.0, 2.5, 5.0, 10.0, 25.0, 50.0, 100.0) .buckets(2.5, 10.0, 25.0, 100.0)
.create(); .create();
} }
@@ -68,7 +77,7 @@ public class PrometheusClientInstanceProfiler implements ClientInstanceProfiler
errorsCounter.labels(destination, Short.toString(errorCode)).inc(); errorsCounter.labels(destination, Short.toString(errorCode)).inc();
} }
long latency = result.getLatency(); long latency = result.getLatency();
responseLatency.labels(destination).observe(latency / 1000000); responseLatency.labels(destination).observe(((double) latency) / NANO_PER_MILLI);
switch (type) { switch (type) {
case GET: case GET:
boolean empty = result.getEmpty(); boolean empty = result.getEmpty();

6
prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/SinkCollector.java
View File

@@ -18,6 +18,7 @@ import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicLong;
import static com.alibaba.otter.canal.prometheus.CanalInstanceExports.DEST_LABELS_LIST; import static com.alibaba.otter.canal.prometheus.CanalInstanceExports.DEST_LABELS_LIST;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
/** /**
* @author Chuanyi Li * @author Chuanyi Li
@@ -25,8 +26,9 @@ import static com.alibaba.otter.canal.prometheus.CanalInstanceExports.DEST_LABEL
public class SinkCollector extends Collector implements InstanceRegistry { public class SinkCollector extends Collector implements InstanceRegistry {
private static final Logger logger = LoggerFactory.getLogger(SinkCollector.class); private static final Logger logger = LoggerFactory.getLogger(SinkCollector.class);
private static final long NANO_PER_MILLI = 1000 * 1000L;
private static final String SINK_BLOCKING_TIME = "canal_instance_sink_blocking_time"; private static final String SINK_BLOCKING_TIME = "canal_instance_sink_blocking_time";
private static final String SINK_BLOCK_TIME_HELP = "Total sink blocking time"; private static final String SINK_BLOCK_TIME_HELP = "Total sink blocking time in milliseconds";
private final ConcurrentMap<String, SinkMetricsHolder> instances = new ConcurrentHashMap<String, SinkMetricsHolder>(); private final ConcurrentMap<String, SinkMetricsHolder> instances = new ConcurrentHashMap<String, SinkMetricsHolder>();
private SinkCollector() {} private SinkCollector() {}
@@ -45,7 +47,7 @@ public class SinkCollector extends Collector implements InstanceRegistry {
CounterMetricFamily blockingCounter = new CounterMetricFamily(SINK_BLOCKING_TIME, CounterMetricFamily blockingCounter = new CounterMetricFamily(SINK_BLOCKING_TIME,
SINK_BLOCK_TIME_HELP, DEST_LABELS_LIST); SINK_BLOCK_TIME_HELP, DEST_LABELS_LIST);
for (SinkMetricsHolder smh : instances.values()) { for (SinkMetricsHolder smh : instances.values()) {
blockingCounter.addMetric(smh.destLabelValues, smh.eventsSinkBlockingTime.doubleValue()); blockingCounter.addMetric(smh.destLabelValues, (smh.eventsSinkBlockingTime.doubleValue() / NANO_PER_MILLI));
} }
mfs.add(blockingCounter); mfs.add(blockingCounter);
return mfs; return mfs;

5
sink/src/main/java/com/alibaba/otter/canal/sink/entry/EntryEventSink.java
View File

@@ -164,6 +164,11 @@ public class EntryEventSink extends AbstractCanalEventSink<List<CanalEntry.Entry
blockingStart = System.nanoTime(); blockingStart = System.nanoTime();
} }
applyWait(++fullTimes); applyWait(++fullTimes);
if (fullTimes % 1000 == 0) {
long nextStart = System.nanoTime();
eventsSinkBlockingTime.addAndGet(nextStart - blockingStart);
blockingStart = nextStart;
}
} }
for (CanalEventDownStreamHandler<List<Event>> handler : getHandlers()) { for (CanalEventDownStreamHandler<List<Event>> handler : getHandlers()) {