github/DataX
5
0
Fork 0
mirror of https://github.com/alibaba/DataX.git synced 2025-05-03 23:09:38 +08:00
Code Issues Packages Projects Releases Wiki Activity

修复统计数量不正确的问题。

Browse Source 
问题原因：终止task的时，会调用this.channel.pushTerminate(TerminateRecord.get())，push一个没有元素的TerminateRecord到queue中，所以pull时，每有1个task统计就会多1个。
在入库时，TerminateRecord会被排除掉，但是在统计时没有处理，引起了该问题。
This commit is contained in:
chenyang 2020-12-09 11:38:13 +08:00
parent 3f9e236329
commit 4df71b8c2d
1 changed files with 8 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
core/src/main/java/com/alibaba/datax/core/transport/channel/Channel.java
View File

@ -141,13 +141,20 @@ public abstract class Channel {
public Record pull() { public Record pull() {
Record record = this.doPull(); Record record = this.doPull();
// record为终止对象时，该记录不算入统计数量
if (!(record instanceof TerminateRecord)) {
this.statPull(1L, record.getByteSize()); this.statPull(1L, record.getByteSize());
}
return record; return record;
} }
public void pullAll(final Collection<Record> rs) { public void pullAll(final Collection<Record> rs) {
Validate.notNull(rs); Validate.notNull(rs);
this.doPullAll(rs); this.doPullAll(rs);
// record为终止对象时，该记录不算入统计数量
long recordSize = rs.stream().filter(record -> !(record instanceof TerminateRecord)).count();
this.statPull(rs.size(), this.getByteSize(rs)); this.statPull(rs.size(), this.getByteSize(rs));
} }