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

fixed issue #1897 , memory event store ack null

Browse Source
This commit is contained in:
agapple
2019-06-25 17:32:46 +08:00
parent adccc0e912
commit 4957083493
6 changed files with 50 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
store/src/main/java/com/alibaba/otter/canal/store/CanalEventStore.java
View File

@@ -76,6 +76,13 @@ public interface CanalEventStore<T> extends CanalLifeCycle, CanalStoreScavenge {
*/
void ack(Position position) throws CanalStoreException;
/**
* 删除指定seqId之前的数据
*
* @Since 1.1.4
*/
void ack(Position position, Long seqId) throws CanalStoreException;
/**
* 出错时执行回滚操作(未提交ack的所有状态信息重新归位，减少出错时数据全部重来的成本)
*/

4
store/src/main/java/com/alibaba/otter/canal/store/helper/CanalEventUtils.java
View File

@@ -70,6 +70,10 @@ public class CanalEventUtils {
position.setPosition(event.getPosition());
position.setTimestamp(event.getExecuteTime());
position.setIncluded(included);
// add serverId at 2016-06-28
position.setServerId(event.getServerId());
// add gtid
position.setGtid(event.getGtid());
LogPosition logPosition = new LogPosition();
logPosition.setPostion(position);

32
store/src/main/java/com/alibaba/otter/canal/store/memory/MemoryEventStoreWithBuffer.java
View File

@@ -334,6 +334,7 @@ public class MemoryEventStoreWithBuffer extends AbstractCanalStoreScavenge imple
range.setStart(CanalEventUtils.createPosition(entrys.get(0)));
range.setEnd(CanalEventUtils.createPosition(entrys.get(result.getEvents().size() - 1)));
range.setEndSeq(end);
// 记录一下是否存在可以被ack的点
for (int i = entrys.size() - 1; i >= 0; i--) {
@@ -369,9 +370,9 @@ public class MemoryEventStoreWithBuffer extends AbstractCanalStoreScavenge imple
return CanalEventUtils.createPosition(event, false);
} else if (firstSeqeuence > INIT_SEQUENCE && firstSeqeuence < putSequence.get()) {
// ack未追上put操作
Event event = entries[getIndex(firstSeqeuence + 1)]; // 最后一次ack的位置数据
// + 1
return CanalEventUtils.createPosition(event, true);
Event event = entries[getIndex(firstSeqeuence)]; // 最后一次ack的位置数据,需要移动到下一条,included
// = false
return CanalEventUtils.createPosition(event, false);
} else if (firstSeqeuence > INIT_SEQUENCE && firstSeqeuence == putSequence.get()) {
// 已经追上，store中没有数据
Event event = entries[getIndex(firstSeqeuence)]; // 最后一次ack的位置数据，和last为同一条，included
@@ -410,10 +411,19 @@ public class MemoryEventStoreWithBuffer extends AbstractCanalStoreScavenge imple
}
public void ack(Position position) throws CanalStoreException {
cleanUntil(position);
cleanUntil(position, -1L);
}
public void ack(Position position, Long seqId) throws CanalStoreException {
cleanUntil(position, seqId);
}
@Override
public void cleanUntil(Position position) throws CanalStoreException {
cleanUntil(position, -1L);
}
public void cleanUntil(Position position, Long seqId) throws CanalStoreException {
final ReentrantLock lock = this.lock;
lock.lock();
try {
@@ -425,6 +435,9 @@ public class MemoryEventStoreWithBuffer extends AbstractCanalStoreScavenge imple
// ack没有list，但有已存在的foreach，还是节省一下list的开销
long localExecTime = 0L;
int deltaRows = 0;
if (seqId > 0) {
maxSequence = seqId;
}
for (long next = sequence + 1; next <= maxSequence; next++) {
Event event = entries[getIndex(next)];
if (localExecTime == 0 && event.getExecuteTime() > 0) {
@@ -432,8 +445,8 @@ public class MemoryEventStoreWithBuffer extends AbstractCanalStoreScavenge imple
}
deltaRows += event.getRowsCount();
memsize += calculateSize(event);
boolean match = CanalEventUtils.checkPosition(event, (LogPosition) position);
if (match) {// 找到对应的position，更新ack seq
if ((seqId < 0 || next == seqId) && CanalEventUtils.checkPosition(event, (LogPosition) position)) {
// 找到对应的position，更新ack seq
hasMatch = true;
if (batchMode.isMemSize()) {
@@ -442,6 +455,12 @@ public class MemoryEventStoreWithBuffer extends AbstractCanalStoreScavenge imple
for (long index = sequence + 1; index < next; index++) {
entries[getIndex(index)] = null;// 设置为null
}
// 考虑getFirstPosition/getLastPosition会获取最后一次ack的position信息
// ack清理的时候只处理entry=null，释放内存
Event lastEvent = entries[getIndex(next)];
lastEvent.setEntry(null);
lastEvent.setRawEntry(null);
}
if (ackSequence.compareAndSet(sequence, next)) {// 避免并发ack
@@ -681,5 +700,4 @@ public class MemoryEventStoreWithBuffer extends AbstractCanalStoreScavenge imple
return ackTableRows;
}
}

5
store/src/test/java/com/alibaba/otter/cancel/store/memory/buffer/MemoryEventStoreMemBatchTest.java
View File

@@ -300,10 +300,11 @@ public class MemoryEventStoreMemBatchTest extends MemoryEventStoreBase {
first = eventStore.getFirstPosition();
lastest = eventStore.getLatestPosition();
List<Event> entrys = new ArrayList<Event>(entrys2.getEvents());
Assert.assertEquals(first, entrys2.getPositionRange().getStart());
// Assert.assertEquals(first, entrys2.getPositionRange().getStart());
Assert.assertEquals(lastest, entrys2.getPositionRange().getEnd());
Assert.assertEquals(first, CanalEventUtils.createPosition(entrys.get(0)));
// Assert.assertEquals(first,
// CanalEventUtils.createPosition(entrys.get(0)));
Assert.assertEquals(lastest, CanalEventUtils.createPosition(entrys.get(entrys.size() - 1)));
// 全部ack掉