azexcy commented on code in PR #25787:
URL: https://github.com/apache/shardingsphere/pull/25787#discussion_r1198670941
##########
kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/task/IncrementalTask.java:
##########
@@ -94,24 +93,11 @@ private IncrementalTaskProgress
createIncrementalTaskProgress(final IngestPositi
return result;
}
- private Collection<Importer> createImporters(final int concurrency, final
ImporterConfiguration importerConfig, final ImporterConnector
importerConnector, final PipelineChannel channel,
- final
PipelineJobProgressListener jobProgressListener) {
- Collection<Importer> result = new LinkedList<>();
- for (int i = 0; i < concurrency; i++) {
- result.add(TypedSPILoader.getService(ImporterCreator.class,
importerConnector.getType()).createImporter(importerConfig, importerConnector,
channel, jobProgressListener,
- ImporterType.INCREMENTAL));
- }
- return result;
- }
-
- private PipelineChannel createChannel(final int concurrency, final
PipelineChannelCreator pipelineChannelCreator, final IncrementalTaskProgress
progress) {
- return pipelineChannelCreator.createPipelineChannel(concurrency,
records -> {
+ private PipelineChannel createChannel(final PipelineChannelCreator
pipelineChannelCreator, final IncrementalTaskProgress progress) {
+ return pipelineChannelCreator.createPipelineChannel(1, records -> {
Record lastHandledRecord = records.get(records.size() - 1);
- if (!(lastHandledRecord.getPosition() instanceof
PlaceholderPosition)) {
- progress.setPosition(lastHandledRecord.getPosition());
-
progress.getIncrementalTaskDelay().setLastEventTimestamps(lastHandledRecord.getCommitTime());
- }
-
progress.getIncrementalTaskDelay().setLatestActiveTimeMillis(System.currentTimeMillis());
+ progress.setPosition(lastHandledRecord.getPosition());
+
progress.getIncrementalTaskDelay().setLastEventTimestamps(lastHandledRecord.getCommitTime());
Review Comment:
My misstake, i will rollback it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
