Skip to content
Permalink
Browse files

0002681: Only query incoming_error for batch being retried

  • Loading branch information...
erilong committed Jul 15, 2016
1 parent 6a89f23 commit a9725930164bd5307f878010683b9adc6255bc80
@@ -516,7 +516,8 @@ private void logDataReceivedFromPush(Node sourceNode, List<IncomingBatch> batchL
@Override
protected IDataWriter chooseDataWriter(Batch batch) {
return buildDataWriter(processInfo, sourceNode.getNodeId(),
batch.getChannelId(), batch.getBatchId());
batch.getChannelId(), batch.getBatchId(),
((ManageIncomingBatchListener) listener).getCurrentBatch().isRetry());
}
};
processor.process(ctx);
@@ -573,7 +574,7 @@ protected void logAndRethrow(Node remoteNode, Throwable ex) throws IOException {
}

protected IDataWriter buildDataWriter(ProcessInfo processInfo, String sourceNodeId,
String channelId, long batchId) {
String channelId, long batchId, boolean isRetry) {
TransformTable[] transforms = null;
NodeGroupLink link = null;
List<ResolvedData> resolvedDatas = new ArrayList<ResolvedData>();
@@ -612,15 +613,16 @@ protected IDataWriter buildDataWriter(ProcessInfo processInfo, String sourceNode
transforms = transformsList != null ? transformsList
.toArray(new TransformTable[transformsList.size()]) : null;

List<IncomingError> incomingErrors = getIncomingErrors(batchId, sourceNodeId);
for (IncomingError incomingError : incomingErrors) {
if (incomingError.isResolveIgnore()
|| StringUtils.isNotBlank(incomingError.getResolveData())) {
resolvedDatas.add(new ResolvedData(incomingError.getFailedRowNumber(),
incomingError.getResolveData(), incomingError.isResolveIgnore()));
if (isRetry) {
List<IncomingError> incomingErrors = getIncomingErrors(batchId, sourceNodeId);
for (IncomingError incomingError : incomingErrors) {
if (incomingError.isResolveIgnore()
|| StringUtils.isNotBlank(incomingError.getResolveData())) {
resolvedDatas.add(new ResolvedData(incomingError.getFailedRowNumber(),
incomingError.getResolveData(), incomingError.isResolveIgnore()));
}
}
}

}

TransformWriter transformWriter = new TransformWriter(platform, TransformPoint.LOAD, null,
@@ -922,7 +924,8 @@ public IncomingBatch call() throws Exception {
batch.getTargetNodeId(), resource), null, listener, "data load from stage") {
@Override
protected IDataWriter chooseDataWriter(Batch batch) {
return buildDataWriter(processInfo, sourceNodeId, batch.getChannelId(), batch.getBatchId());
return buildDataWriter(processInfo, sourceNodeId, batch.getChannelId(), batch.getBatchId(),
((ManageIncomingBatchListener) listener).getCurrentBatch().isRetry());
}
};
processor.process(ctx);

0 comments on commit a972593

Please sign in to comment.
You can’t perform that action at this time.