This is an automated email from the ASF dual-hosted git repository.
menghaoran pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shardingsphere.git
The following commit(s) were added to refs/heads/master by this push:
new 3284e66 Simplify MemoryPipelineChannel impl to resolve scaling OOME
on large data set (#14978)
3284e66 is described below
commit 3284e664ceacd142fd4e198214a1dfcf7dbb0796
Author: Hongsheng Zhong <[email protected]>
AuthorDate: Fri Jan 21 16:25:54 2022 +0800
Simplify MemoryPipelineChannel impl to resolve scaling OOME on large data
set (#14978)
* Improve log
* Simplify MemoryPipelineChannel impl
---
.../pipeline/core/importer/AbstractImporter.java | 12 ++-
.../distribution/AbstractBitSetChannel.java | 61 ---------------
.../distribution/AutoAcknowledgeChannel.java | 45 -----------
.../ingest/channel/distribution/BitSetChannel.java | 80 --------------------
.../channel/distribution/BlockingQueueChannel.java | 18 ++---
.../distribution/MemoryPipelineChannel.java | 87 +++-------------------
.../ingest/dumper/AbstractInventoryDumper.java | 15 +++-
.../core/prepare/InventoryTaskSplitter.java | 1 +
.../data/pipeline/core/task/InventoryTask.java | 1 +
.../api/executor/AbstractLifecycleExecutor.java | 3 +
.../AutoAcknowledgePipelineChannelTest.java | 55 --------------
11 files changed, 44 insertions(+), 334 deletions(-)
diff --git
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/importer/AbstractImporter.java
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/importer/AbstractImporter.java
index e7eefb7..62be369 100644
---
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/importer/AbstractImporter.java
+++
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/importer/AbstractImporter.java
@@ -89,22 +89,32 @@ public abstract class AbstractImporter extends
AbstractLifecycleExecutor impleme
@Override
public final void write() {
log.info("importer write");
+ int round = 1;
int rowCount = 0;
+ boolean finishedByBreak = false;
while (isRunning()) {
List<Record> records = channel.fetchRecords(1024, 3);
if (null != records && !records.isEmpty()) {
+ round++;
rowCount += records.size();
flush(dataSourceManager.getDataSource(importerConfig.getDataSourceConfig()),
records);
if (null != importerListener) {
importerListener.recordsImported(records);
}
channel.ack(records);
+ if (log.isDebugEnabled()) {
+ log.debug("importer write, round={}, rowCount={}", round,
rowCount);
+ } else if (0 == round % 50) {
+ log.info("importer write, round={}, rowCount={}", round,
rowCount);
+ }
if (FinishedRecord.class.equals(records.get(records.size() -
1).getClass())) {
+ log.info("write, get FinishedRecord, break");
+ finishedByBreak = true;
break;
}
}
}
- log.info("importer write, rowCount={}", rowCount);
+ log.info("importer write done, rowCount={}, finishedByBreak={}",
rowCount, finishedByBreak);
}
private void flush(final DataSource dataSource, final List<Record> buffer)
{
diff --git
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/channel/distribution/AbstractBitSetChannel.java
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/channel/distribution/AbstractBitSetChannel.java
deleted file mode 100644
index e1cefc3..0000000
---
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/channel/distribution/AbstractBitSetChannel.java
+++ /dev/null
@@ -1,61 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package
org.apache.shardingsphere.data.pipeline.core.ingest.channel.distribution;
-
-import lombok.AccessLevel;
-import lombok.Getter;
-import lombok.Setter;
-import org.apache.shardingsphere.data.pipeline.api.ingest.record.Record;
-
-import java.util.BitSet;
-import java.util.Deque;
-import java.util.concurrent.ConcurrentLinkedDeque;
-
-/**
- * Abstract BitSet channel.
- */
-@Getter(AccessLevel.PROTECTED)
-@Setter(AccessLevel.PROTECTED)
-public abstract class AbstractBitSetChannel implements BitSetChannel {
-
- private final Deque<Record> toBeAckRecords = new ConcurrentLinkedDeque<>();
-
- private final ManualBitSet manualBitSet = new ManualBitSet();
-
- private volatile long acknowledgedIndex;
-
- @Override
- public BitSet getAckBitSet(final long fromIndex) {
- return manualBitSet.get(fromIndex, acknowledgedIndex);
- }
-
- @Override
- public Record removeAckRecord() {
- return toBeAckRecords.remove();
- }
-
- @Override
- public void clear(final long index) {
- manualBitSet.clear(index);
- }
-
- @Override
- public void close() {
- toBeAckRecords.clear();
- }
-}
diff --git
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/channel/distribution/AutoAcknowledgeChannel.java
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/channel/distribution/AutoAcknowledgeChannel.java
deleted file mode 100644
index ea1e508..0000000
---
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/channel/distribution/AutoAcknowledgeChannel.java
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package
org.apache.shardingsphere.data.pipeline.core.ingest.channel.distribution;
-
-import org.apache.shardingsphere.data.pipeline.api.ingest.record.Record;
-
-import java.util.List;
-
-/**
- * Auto Acknowledge BitSet channel.
- */
-public final class AutoAcknowledgeChannel extends AbstractBitSetChannel {
-
- @Override
- public void pushRecord(final Record dataRecord, final long index) {
- getManualBitSet().set(index);
- getToBeAckRecords().add(dataRecord);
- setAcknowledgedIndex(index + 1);
- }
-
- @Override
- public List<Record> fetchRecords(final int batchSize, final int
timeoutSeconds) {
- throw new UnsupportedOperationException("Auto ack channel can not
fetch records.");
- }
-
- @Override
- public void ack(final List<Record> records) {
- throw new UnsupportedOperationException("Auto ack channel do not have
to ack.");
- }
-}
diff --git
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/channel/distribution/BitSetChannel.java
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/channel/distribution/BitSetChannel.java
deleted file mode 100644
index 0ca3b30..0000000
---
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/channel/distribution/BitSetChannel.java
+++ /dev/null
@@ -1,80 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package
org.apache.shardingsphere.data.pipeline.core.ingest.channel.distribution;
-
-import org.apache.shardingsphere.data.pipeline.api.ingest.record.Record;
-
-import java.util.BitSet;
-import java.util.List;
-
-/**
- * BitSet channel interface.
- */
-public interface BitSetChannel {
-
- /**
- * Push a {@code DataRecord} with index to channel.
- *
- * @param dataRecord data
- * @param index data index
- */
- void pushRecord(Record dataRecord, long index);
-
- /**
- * Fetch {@code Record} from channel, if the timeout also returns the
record.
- *
- * @param batchSize record batch size
- * @param timeoutSeconds timeout(seconds)
- * @return record
- */
- List<Record> fetchRecords(int batchSize, int timeoutSeconds);
-
- /**
- * Ack the last batch.
- *
- * @param records record list
- */
- void ack(List<Record> records);
-
- /**
- * Get acknowledged BitSet.
- *
- * @param fromIndex from index
- * @return BitSet
- */
- BitSet getAckBitSet(long fromIndex);
-
- /**
- * Remove earliest acknowledged record.
- *
- * @return record
- */
- Record removeAckRecord();
-
- /**
- * Clear BitSet.
- *
- * @param index BitSet index
- */
- void clear(long index);
-
- /**
- * Close channel.
- */
- void close();
-}
diff --git
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/channel/distribution/BlockingQueueChannel.java
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/channel/distribution/BlockingQueueChannel.java
index e88cebd..3db7b03 100644
---
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/channel/distribution/BlockingQueueChannel.java
+++
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/channel/distribution/BlockingQueueChannel.java
@@ -17,6 +17,7 @@
package
org.apache.shardingsphere.data.pipeline.core.ingest.channel.distribution;
+import
org.apache.shardingsphere.data.pipeline.api.ingest.channel.PipelineChannel;
import org.apache.shardingsphere.data.pipeline.api.ingest.record.Record;
import org.apache.shardingsphere.data.pipeline.core.util.ThreadUtil;
@@ -26,14 +27,13 @@ import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
/**
- * Blocking queue BitSet channel.
+ * Blocking queue pipeline channel.
*/
-public final class BlockingQueueChannel extends AbstractBitSetChannel {
+// TODO rename
+public final class BlockingQueueChannel implements PipelineChannel {
private final BlockingQueue<Record> queue;
- private long fetchedIndex;
-
public BlockingQueueChannel() {
this(10000);
}
@@ -43,12 +43,11 @@ public final class BlockingQueueChannel extends
AbstractBitSetChannel {
}
@Override
- public void pushRecord(final Record dataRecord, final long index) {
- getManualBitSet().set(index);
+ public void pushRecord(final Record dataRecord) {
try {
queue.put(dataRecord);
} catch (final InterruptedException ex) {
- throw new RuntimeException("put " + dataRecord + " into queue at
index " + index + " failed", ex);
+ throw new RuntimeException("put " + dataRecord + " into queue
failed", ex);
}
}
@@ -64,20 +63,15 @@ public final class BlockingQueueChannel extends
AbstractBitSetChannel {
ThreadUtil.sleep(100L);
}
queue.drainTo(result, batchSize);
- // TODO memory released after job completed?
- getToBeAckRecords().addAll(result);
- fetchedIndex = getManualBitSet().getEndIndex(fetchedIndex,
result.size());
return result;
}
@Override
public void ack(final List<Record> records) {
- setAcknowledgedIndex(fetchedIndex);
}
@Override
public void close() {
queue.clear();
- super.close();
}
}
diff --git
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/channel/distribution/MemoryPipelineChannel.java
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/channel/distribution/MemoryPipelineChannel.java
index b845ae9..1667299 100644
---
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/channel/distribution/MemoryPipelineChannel.java
+++
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/channel/distribution/MemoryPipelineChannel.java
@@ -25,43 +25,25 @@ import
org.apache.shardingsphere.data.pipeline.api.ingest.record.FinishedRecord;
import
org.apache.shardingsphere.data.pipeline.api.ingest.record.PlaceholderRecord;
import org.apache.shardingsphere.data.pipeline.api.ingest.record.Record;
-import java.util.BitSet;
import java.util.HashMap;
-import java.util.LinkedList;
import java.util.List;
import java.util.Map;
-import java.util.Queue;
-import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
/**
- * Distribution channel.
+ * Memory pipeline channel.
*/
@Slf4j
+// TODO rename
public final class MemoryPipelineChannel implements PipelineChannel {
private final int channelNumber;
- private final BitSetChannel[] channels;
-
- // TODO remove autoAckChannel
- private final BitSetChannel autoAckChannel = new AutoAcknowledgeChannel();
+ private final PipelineChannel[] channels;
private final Map<String, Integer> channelAssignment = new HashMap<>();
private final AckCallback ackCallback;
- private final AtomicLong indexAutoIncreaseGenerator = new AtomicLong();
-
- private final Queue<Integer> toBeAckBitSetIndexes = new
ConcurrentLinkedQueue<>();
-
- private long lastAckIndex;
-
- private ScheduledExecutorService scheduleAckRecordsExecutor;
-
public MemoryPipelineChannel(final AckCallback ackCallback) {
this(10000, ackCallback);
}
@@ -73,23 +55,16 @@ public final class MemoryPipelineChannel implements
PipelineChannel {
public MemoryPipelineChannel(final int channelNumber, final int
blockQueueSize, final AckCallback ackCallback) {
this.channelNumber = channelNumber;
this.ackCallback = ackCallback;
- channels = new BitSetChannel[channelNumber];
+ channels = new PipelineChannel[channelNumber];
for (int i = 0; i < channelNumber; i++) {
channels[i] = new BlockingQueueChannel(blockQueueSize);
}
- scheduleAckRecords();
- }
-
- // TODO remove scheduleAckRecords
- private void scheduleAckRecords() {
- scheduleAckRecordsExecutor =
Executors.newSingleThreadScheduledExecutor();
- scheduleAckRecordsExecutor.scheduleWithFixedDelay(this::ackRecords0,
5, 1, TimeUnit.SECONDS);
}
@Override
public void pushRecord(final Record record) {
if (FinishedRecord.class.equals(record.getClass())) {
- for (int i = 0; i < channels.length; i++) {
+ for (int i = 0; i < channelNumber; i++) {
pushRecord(record, i);
}
} else if (DataRecord.class.equals(record.getClass())) {
@@ -101,9 +76,9 @@ public final class MemoryPipelineChannel implements
PipelineChannel {
}
}
- private void pushRecord(final Record record, final int index) {
- toBeAckBitSetIndexes.add(index);
- getBitSetChannel(index).pushRecord(record,
indexAutoIncreaseGenerator.getAndIncrement());
+ private void pushRecord(final Record record, final int channelIndex) {
+ PipelineChannel channel = channels[channelIndex];
+ channel.pushRecord(record);
}
@Override
@@ -117,45 +92,7 @@ public final class MemoryPipelineChannel implements
PipelineChannel {
ackCallback.onAck(records);
}
- private synchronized void ackRecords0() {
- try {
- int count = shouldAckCount();
- if (0 == count) {
- return;
- }
- //ackCallback.onAck(fetchAckRecords(count));
- lastAckIndex += count;
- for (BitSetChannel channel : channels) {
- channel.clear(lastAckIndex);
- }
- // CHECKSTYLE:OFF
- } catch (final Exception ex) {
- // CHECKSTYLE:ON
- log.error("distribution channel auto ack failed.", ex);
- }
- }
-
- private int shouldAckCount() {
- BitSet bitSet = autoAckChannel.getAckBitSet(lastAckIndex);
- for (BitSetChannel channel : channels) {
- bitSet.or(channel.getAckBitSet(lastAckIndex));
- }
- return bitSet.nextClearBit(0);
- }
-
- private List<Record> fetchAckRecords(final int count) {
- List<Record> result = new LinkedList<>();
- for (int i = 0; i < count; i++) {
-
result.add(getBitSetChannel(toBeAckBitSetIndexes.remove()).removeAckRecord());
- }
- return result;
- }
-
- private BitSetChannel getBitSetChannel(final Integer index) {
- return index == -1 ? autoAckChannel : channels[index];
- }
-
- private BitSetChannel findChannel() {
+ private PipelineChannel findChannel() {
String threadId = Long.toString(Thread.currentThread().getId());
checkAssignment(threadId);
return channels[channelAssignment.get(threadId)];
@@ -182,12 +119,8 @@ public final class MemoryPipelineChannel implements
PipelineChannel {
@Override
public void close() {
- // TODO shutdownNow?
- scheduleAckRecordsExecutor.shutdown();
- ackRecords0();
- for (BitSetChannel each : channels) {
+ for (PipelineChannel each : channels) {
each.close();
}
- toBeAckBitSetIndexes.clear();
}
}
diff --git
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/dumper/AbstractInventoryDumper.java
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/dumper/AbstractInventoryDumper.java
index 7d44841..8fb9f84 100644
---
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/dumper/AbstractInventoryDumper.java
+++
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/dumper/AbstractInventoryDumper.java
@@ -98,18 +98,23 @@ public abstract class AbstractInventoryDumper extends
AbstractLifecycleExecutor
String sql = getDumpSQL();
IngestPosition<?> position = inventoryDumperConfig.getPosition();
log.info("inventory dump, sql={}, position={}", sql, position);
+ // TODO [ksdfjaskjiowetr] handle FinishedPosition
try (Connection conn =
dataSourceManager.getDataSource(inventoryDumperConfig.getDataSourceConfig()).getConnection())
{
+ int round = 1;
Number startUniqueKeyValue = getPositionBeginValue(position) - 1;
Optional<Number> maxUniqueKeyValue;
- while ((maxUniqueKeyValue = dump0(conn, sql,
startUniqueKeyValue)).isPresent()) {
+ while ((maxUniqueKeyValue = dump0(conn, sql, startUniqueKeyValue,
round++)).isPresent()) {
startUniqueKeyValue = maxUniqueKeyValue.get();
}
+ log.info("inventory dump done, round={}, maxUniqueKeyValue={}",
round, maxUniqueKeyValue);
} catch (final SQLException ex) {
+ log.error("inventory dump, ex caught, msg={}", ex.getMessage());
stop();
// TODO channel.close() when job success too, e.g.
InventoryTask/IncrementalTask?
channel.close();
throw new IngestException(ex);
} finally {
+ log.info("inventory dump, before put FinishedRecord");
pushRecord(new FinishedRecord(new FinishedPosition()));
}
}
@@ -120,7 +125,7 @@ public abstract class AbstractInventoryDumper extends
AbstractLifecycleExecutor
return "SELECT * FROM " + tableName + " WHERE " + primaryKey + " > ?
AND " + primaryKey + " <= ? ORDER BY " + primaryKey + " ASC LIMIT ?";
}
- private Optional<Number> dump0(final Connection conn, final String sql,
final Number startUniqueKeyValue) throws SQLException {
+ private Optional<Number> dump0(final Connection conn, final String sql,
final Number startUniqueKeyValue, final int round) throws SQLException {
if (null != rateLimitAlgorithm) {
rateLimitAlgorithm.intercept(JobOperationType.SELECT, 1);
}
@@ -147,7 +152,11 @@ public abstract class AbstractInventoryDumper extends
AbstractLifecycleExecutor
pushRecord(record);
rowCount++;
}
- log.info("dump, rowCount={}, maxUniqueKeyValue={}", rowCount,
maxUniqueKeyValue);
+ if (log.isDebugEnabled()) {
+ log.debug("dump, round={}, rowCount={},
maxUniqueKeyValue={}", round, rowCount, maxUniqueKeyValue);
+ } else if (0 == round % 50) {
+ log.info("dump, round={}, rowCount={},
maxUniqueKeyValue={}", round, rowCount, maxUniqueKeyValue);
+ }
return Optional.ofNullable(maxUniqueKeyValue);
}
}
diff --git
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/prepare/InventoryTaskSplitter.java
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/prepare/InventoryTaskSplitter.java
index 0b2cd1a..46c7d23 100644
---
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/prepare/InventoryTaskSplitter.java
+++
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/prepare/InventoryTaskSplitter.java
@@ -131,6 +131,7 @@ public final class InventoryTaskSplitter {
JobProgress initProgress = jobContext.getInitProgress();
if (null != initProgress && initProgress.getStatus() !=
JobStatus.PREPARING_FAILURE) {
Collection<IngestPosition<?>> result =
jobContext.getInitProgress().getInventoryPosition(dumperConfig.getTableName()).values();
+ // TODO [ksdfjaskjiowetr] seems findFirst() might cause issue when
first one is not PrimaryKeyPosition
result.stream().findFirst().ifPresent(position -> {
if (position instanceof PrimaryKeyPosition) {
String primaryKey =
metaDataManager.getTableMetaData(dumperConfig.getTableName(),
databaseType).getPrimaryKeyColumns().get(0);
diff --git
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/task/InventoryTask.java
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/task/InventoryTask.java
index 313c686..4f68fea 100644
---
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/task/InventoryTask.java
+++
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/task/InventoryTask.java
@@ -112,6 +112,7 @@ public final class InventoryTask extends
AbstractLifecycleExecutor implements Pi
private void instanceChannel(final Importer importer) {
PipelineChannel channel =
pipelineChannelFactory.createPipelineChannel(1, records -> {
+ // TODO find in reversed order
Optional<Record> record = records.stream().filter(each ->
!(each.getPosition() instanceof PlaceholderPosition)).reduce((a, b) -> b);
record.ifPresent(value -> position = value.getPosition());
});
diff --git
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-spi/src/main/java/org/apache/shardingsphere/data/pipeline/api/executor/AbstractLifecycleExecutor.java
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-spi/src/main/java/org/apache/shardingsphere/data/pipeline/api/executor/AbstractLifecycleExecutor.java
index 6e1649a..e358bf7 100644
---
a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-spi/src/main/java/org/apache/shardingsphere/data/pipeline/api/executor/AbstractLifecycleExecutor.java
+++
b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-spi/src/main/java/org/apache/shardingsphere/data/pipeline/api/executor/AbstractLifecycleExecutor.java
@@ -20,12 +20,14 @@ package
org.apache.shardingsphere.data.pipeline.api.executor;
import lombok.AccessLevel;
import lombok.Getter;
import lombok.Setter;
+import lombok.extern.slf4j.Slf4j;
/**
* Abstract lifecycle executor.
*/
@Getter
@Setter
+@Slf4j
public abstract class AbstractLifecycleExecutor implements LifecycleExecutor {
@Setter(AccessLevel.PROTECTED)
@@ -39,6 +41,7 @@ public abstract class AbstractLifecycleExecutor implements
LifecycleExecutor {
@Override
public void stop() {
+ log.info("stop lifecycle executor: {}", super.toString());
running = false;
}
diff --git
a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/ingest/channel/distribution/AutoAcknowledgePipelineChannelTest.java
b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/ingest/channel/distribution/AutoAcknowledgePipelineChannelTest.java
deleted file mode 100644
index e9de710..0000000
---
a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/ingest/channel/distribution/AutoAcknowledgePipelineChannelTest.java
+++ /dev/null
@@ -1,55 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package
org.apache.shardingsphere.data.pipeline.core.ingest.channel.distribution;
-
-import
org.apache.shardingsphere.data.pipeline.api.ingest.position.PlaceholderPosition;
-import org.apache.shardingsphere.data.pipeline.api.ingest.record.DataRecord;
-import org.junit.Before;
-import org.junit.Test;
-
-import java.util.BitSet;
-import java.util.Collections;
-
-import static org.junit.Assert.assertTrue;
-
-public final class AutoAcknowledgePipelineChannelTest {
-
- private AutoAcknowledgeChannel channel;
-
- @Before
- public void setUp() {
- channel = new AutoAcknowledgeChannel();
- }
-
- @Test
- public void assertPushRecord() {
- channel.pushRecord(new DataRecord(new PlaceholderPosition(), 1), 0);
- BitSet bitSet = channel.getAckBitSet(0);
- assertTrue(bitSet.get(0));
- }
-
- @Test(expected = UnsupportedOperationException.class)
- public void assertFetchRecordsFailure() {
- channel.fetchRecords(1, 1);
- }
-
- @Test(expected = UnsupportedOperationException.class)
- public void assertAckFailure() {
- channel.ack(Collections.emptyList());
- }
-}
