mirror of
https://github.com/alibaba/canal.git
synced 2026-11-04 00:30:15 +08:00
@@ -220,8 +220,8 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
|
|||||||
events = getEvents(canalInstance.getEventStore(), start, batchSize, timeout, unit);
|
events = getEvents(canalInstance.getEventStore(), start, batchSize, timeout, unit);
|
||||||
|
|
||||||
if (CollectionUtils.isEmpty(events.getEvents())) {
|
if (CollectionUtils.isEmpty(events.getEvents())) {
|
||||||
logger.debug("get successfully, clientId:{} batchSize:{} but result is null", new Object[] {
|
logger.debug("get successfully, clientId:{} batchSize:{} but result is null",
|
||||||
clientIdentity.getClientId(), batchSize });
|
clientIdentity.getClientId(), batchSize);
|
||||||
return new Message(-1, new ArrayList<Entry>()); // 返回空包,避免生成batchId,浪费性能
|
return new Message(-1, new ArrayList<Entry>()); // 返回空包,避免生成batchId,浪费性能
|
||||||
} else {
|
} else {
|
||||||
// 记录到流式信息
|
// 记录到流式信息
|
||||||
@@ -232,13 +232,14 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
|
|||||||
return input.getEntry();
|
return input.getEntry();
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
if (logger.isInfoEnabled()) {
|
||||||
logger.info("get successfully, clientId:{} batchSize:{} real size is {} and result is [batchId:{} , position:{}]",
|
logger.info("get successfully, clientId:{} batchSize:{} real size is {} and result is [batchId:{} , position:{}]",
|
||||||
clientIdentity.getClientId(),
|
clientIdentity.getClientId(),
|
||||||
batchSize,
|
batchSize,
|
||||||
entrys.size(),
|
entrys.size(),
|
||||||
batchId,
|
batchId,
|
||||||
events.getPositionRange());
|
events.getPositionRange());
|
||||||
|
}
|
||||||
// 直接提交ack
|
// 直接提交ack
|
||||||
ack(clientIdentity, batchId);
|
ack(clientIdentity, batchId);
|
||||||
return new Message(batchId, entrys);
|
return new Message(batchId, entrys);
|
||||||
@@ -297,8 +298,8 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
|
|||||||
}
|
}
|
||||||
|
|
||||||
if (CollectionUtils.isEmpty(events.getEvents())) {
|
if (CollectionUtils.isEmpty(events.getEvents())) {
|
||||||
logger.debug("getWithoutAck successfully, clientId:{} batchSize:{} but result is null", new Object[] {
|
logger.debug("getWithoutAck successfully, clientId:{} batchSize:{} but result is null",
|
||||||
clientIdentity.getClientId(), batchSize });
|
clientIdentity.getClientId(), batchSize);
|
||||||
return new Message(-1, new ArrayList<Entry>()); // 返回空包,避免生成batchId,浪费性能
|
return new Message(-1, new ArrayList<Entry>()); // 返回空包,避免生成batchId,浪费性能
|
||||||
} else {
|
} else {
|
||||||
// 记录到流式信息
|
// 记录到流式信息
|
||||||
@@ -309,13 +310,14 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
|
|||||||
return input.getEntry();
|
return input.getEntry();
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
if (logger.isInfoEnabled()) {
|
||||||
logger.info("getWithoutAck successfully, clientId:{} batchSize:{} real size is {} and result is [batchId:{} , position:{}]",
|
logger.info("getWithoutAck successfully, clientId:{} batchSize:{} real size is {} and result is [batchId:{} , position:{}]",
|
||||||
clientIdentity.getClientId(),
|
clientIdentity.getClientId(),
|
||||||
batchSize,
|
batchSize,
|
||||||
entrys.size(),
|
entrys.size(),
|
||||||
batchId,
|
batchId,
|
||||||
events.getPositionRange());
|
events.getPositionRange());
|
||||||
|
}
|
||||||
return new Message(batchId, entrys);
|
return new Message(batchId, entrys);
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -377,10 +379,12 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
|
|||||||
// 更新cursor
|
// 更新cursor
|
||||||
if (positionRanges.getAck() != null) {
|
if (positionRanges.getAck() != null) {
|
||||||
canalInstance.getMetaManager().updateCursor(clientIdentity, positionRanges.getAck());
|
canalInstance.getMetaManager().updateCursor(clientIdentity, positionRanges.getAck());
|
||||||
logger.info("ack successfully, clientId:{} batchId:{} position:{}",
|
if (logger.isInfoEnabled()) {
|
||||||
clientIdentity.getClientId(),
|
logger.info("ack successfully, clientId:{} batchId:{} position:{}",
|
||||||
batchId,
|
clientIdentity.getClientId(),
|
||||||
positionRanges);
|
batchId,
|
||||||
|
positionRanges);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// 可定时清理数据
|
// 可定时清理数据
|
||||||
|
|||||||
Reference in New Issue
Block a user