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

Merge b8506be693 into 0824b45c5e

Browse Source
This commit is contained in:
ggzone 2025-04-10 16:23:56 +08:00 committed by GitHub
commit cf6cebd26b
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 12 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -141,15 +141,25 @@ public abstract class Channel {
public Record pull() { public Record pull() {
Record record = this.doPull(); Record record = this.doPull();
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);
if (rs.size() > 0) {
Record record = (Record) rs.toArray()[rs.size() - 1];
//TerminateRecord的bytesize=0，无需特殊处理
if (record instanceof TerminateRecord) {
this.statPull(rs.size() - 1, this.getByteSize(rs));
} else{
this.statPull(rs.size(), this.getByteSize(rs)); this.statPull(rs.size(), this.getByteSize(rs));
} }
}
}
protected abstract void doPush(Record r); protected abstract void doPush(Record r);