Pipe: handle OUT_OF_TTL status code when data syncing

[SteveYurongSu]

As title.

create pipe test11 with source ('source.path'='root.**', 'inclusion'='data,schema' ,'source.realtime.mode'='stream','source.realtime.enable'='true','source.forwarding-pipe-requests'='false','source.batch.enable'='true','source.history.enable'='true') with sink ('sink'='iotdb-thrift-sink', 'sink.node-urls'='iotdb-16:6667');

CREATE DATABASE root.sg_ttl_2;
create timeseries root.sg_ttl_2.dev.status with datatype=double,encoding=PLAIN ;
set ttl to root.sg_ttl_2 5000;
insert into root.sg_ttl_2.dev(time,status) values(1000,2.3);

2024-04-25 11:43:51,672 [pool-37-IoTDB-Pipe-Connector-Executor-Pool-2] ERROR o.a.i.c.c.t.WrappedThreadPoolExecutor:129 - Exception in thread pool org.apache.iotdb.threadpool:type=Pipe-Connector-Executor-Pool 
org.apache.iotdb.pipe.api.exception.PipeConnectionException: PipeConnector: org.apache.iotdb.db.pipe.connector.protocol.thrift.async.IoTDBDataRegionAsyncConnector heartbeat failed, or encountered failure when transferring generic event. Failure: Failed to transfer tablet insertion event PipeInsertNodeTabletInsertionEvent{walEntryHandler=WALEntryHandler{memTableId=2, value=null, walEntryPosition=org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALEntryPosition@1f4224a6}, progressIndex=IoTProgressIndex{peerId2SearchIndex={1=3}}, isAligned=false, isGeneratedByPipe=false, dataContainer=null} - EnrichedEvent{referenceCount=1, isReleased=false, pipeName='test11', pipeTaskMeta=PipeTask{progressIndex='IoTProgressIndex{peerId2SearchIndex={1=2}}', leaderNodeId=1, exceptionMessages='{}'}, committerKey='test11_5_1714016028405', commitId=5, pattern='IoTDBPipePattern{pattern='root.**'}', startTime=-9223372036854775808, endTime=9223372036854775807, isPatternParsed=true, isTimeParsed=true, shouldReportOnCommit=true}, because Transfer PipeInsertNodeTabletInsertionEvent PipeInsertNodeTabletInsertionEvent{walEntryHandler=WALEntryHandler{memTableId=2, value=null, walEntryPosition=org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALEntryPosition@1f4224a6}, progressIndex=IoTProgressIndex{peerId2SearchIndex={1=3}}, isAligned=false, isGeneratedByPipe=false, dataContainer=null} - EnrichedEvent{referenceCount=2, isReleased=false, pipeName='test11', pipeTaskMeta=PipeTask{progressIndex='IoTProgressIndex{peerId2SearchIndex={1=2}}', leaderNodeId=1, exceptionMessages='{}'}, committerKey='test11_5_1714016028405', commitId=5, pattern='IoTDBPipePattern{pattern='root.**'}', startTime=-9223372036854775808, endTime=9223372036854775807, isPatternParsed=true, isTimeParsed=true, shouldReportOnCommit=true} error, result status TSStatus(code:607, message:Insertion time [2024-04-25T11:39:58.560] is less than ttl time bound [2024-04-25T11:43:44.356]).
	at org.apache.iotdb.db.pipe.task.subtask.connector.PipeConnectorSubtask.transferHeartbeatEvent(PipeConnectorSubtask.java:156)
	at org.apache.iotdb.db.pipe.task.subtask.connector.PipeConnectorSubtask.executeOnce(PipeConnectorSubtask.java:108)
	at org.apache.iotdb.commons.pipe.task.subtask.PipeSubtask.call(PipeSubtask.java:80)
	at org.apache.iotdb.commons.pipe.task.subtask.PipeAbstractConnectorSubtask.call(PipeAbstractConnectorSubtask.java:71)
	at org.apache.iotdb.commons.pipe.task.subtask.PipeAbstractConnectorSubtask.call(PipeAbstractConnectorSubtask.java:38)
	at com.google.common.util.concurrent.TrustedListenableFutureTask$TrustedFutureInterruptibleTask.runInterruptibly(TrustedListenableFutureTask.java:131)
	at com.google.common.util.concurrent.InterruptibleTask.run(InterruptibleTask.java:75)
	at com.google.common.util.concurrent.TrustedListenableFutureTask.run(TrustedListenableFutureTask.java:82)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.iotdb.pipe.api.exception.PipeConnectionException: Failed to transfer tablet insertion event PipeInsertNodeTabletInsertionEvent{walEntryHandler=WALEntryHandler{memTableId=2, value=null, walEntryPosition=org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALEntryPosition@1f4224a6}, progressIndex=IoTProgressIndex{peerId2SearchIndex={1=3}}, isAligned=false, isGeneratedByPipe=false, dataContainer=null} - EnrichedEvent{referenceCount=1, isReleased=false, pipeName='test11', pipeTaskMeta=PipeTask{progressIndex='IoTProgressIndex{peerId2SearchIndex={1=2}}', leaderNodeId=1, exceptionMessages='{}'}, committerKey='test11_5_1714016028405', commitId=5, pattern='IoTDBPipePattern{pattern='root.**'}', startTime=-9223372036854775808, endTime=9223372036854775807, isPatternParsed=true, isTimeParsed=true, shouldReportOnCommit=true}, because Transfer PipeInsertNodeTabletInsertionEvent PipeInsertNodeTabletInsertionEvent{walEntryHandler=WALEntryHandler{memTableId=2, value=null, walEntryPosition=org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALEntryPosition@1f4224a6}, progressIndex=IoTProgressIndex{peerId2SearchIndex={1=3}}, isAligned=false, isGeneratedByPipe=false, dataContainer=null} - EnrichedEvent{referenceCount=2, isReleased=false, pipeName='test11', pipeTaskMeta=PipeTask{progressIndex='IoTProgressIndex{peerId2SearchIndex={1=2}}', leaderNodeId=1, exceptionMessages='{}'}, committerKey='test11_5_1714016028405', commitId=5, pattern='IoTDBPipePattern{pattern='root.**'}', startTime=-9223372036854775808, endTime=9223372036854775807, isPatternParsed=true, isTimeParsed=true, shouldReportOnCommit=true} error, result status TSStatus(code:607, message:Insertion time [2024-04-25T11:39:58.560] is less than ttl time bound [2024-04-25T11:43:44.356]).
	at org.apache.iotdb.db.pipe.connector.protocol.thrift.sync.IoTDBDataRegionSyncConnector.transfer(IoTDBDataRegionSyncConnector.java:113)
	at org.apache.iotdb.db.pipe.connector.protocol.thrift.async.IoTDBDataRegionAsyncConnector.transferQueuedEventsIfNecessary(IoTDBDataRegionAsyncConnector.java:349)
	at org.apache.iotdb.db.pipe.connector.protocol.thrift.async.IoTDBDataRegionAsyncConnector.transfer(IoTDBDataRegionAsyncConnector.java:303)
	at org.apache.iotdb.db.pipe.task.subtask.connector.PipeConnectorSubtask.transferHeartbeatEvent(PipeConnectorSubtask.java:150)
	... 10 common frames omitted
Caused by: org.apache.iotdb.commons.exception.pipe.PipeRuntimeConnectorRetryTimesConfigurableException: Transfer PipeInsertNodeTabletInsertionEvent PipeInsertNodeTabletInsertionEvent{walEntryHandler=WALEntryHandler{memTableId=2, value=null, walEntryPosition=org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALEntryPosition@1f4224a6}, progressIndex=IoTProgressIndex{peerId2SearchIndex={1=3}}, isAligned=false, isGeneratedByPipe=false, dataContainer=null} - EnrichedEvent{referenceCount=2, isReleased=false, pipeName='test11', pipeTaskMeta=PipeTask{progressIndex='IoTProgressIndex{peerId2SearchIndex={1=2}}', leaderNodeId=1, exceptionMessages='{}'}, committerKey='test11_5_1714016028405', commitId=5, pattern='IoTDBPipePattern{pattern='root.**'}', startTime=-9223372036854775808, endTime=9223372036854775807, isPatternParsed=true, isTimeParsed=true, shouldReportOnCommit=true} error, result status TSStatus(code:607, message:Insertion time [2024-04-25T11:39:58.560] is less than ttl time bound [2024-04-25T11:43:44.356])
	at org.apache.iotdb.commons.pipe.connector.PipeReceiverStatusHandler.handle(PipeReceiverStatusHandler.java:185)
	at org.apache.iotdb.db.pipe.connector.protocol.thrift.sync.IoTDBDataRegionSyncConnector.doTransfer(IoTDBDataRegionSyncConnector.java:247)
	at org.apache.iotdb.db.pipe.connector.protocol.thrift.sync.IoTDBDataRegionSyncConnector.doTransferWrapper(IoTDBDataRegionSyncConnector.java:200)
	at org.apache.iotdb.db.pipe.connector.protocol.thrift.sync.IoTDBDataRegionSyncConnector.transfer(IoTDBDataRegionSyncConnector.java:106)
	... 13 common frames omitted
2024-04-25 11:43:51,672 [pool-35-IoTDB-Pipe-SubTask-Callback-Executor-Pool-1] WARN  o.a.i.c.p.t.s.PipeAbstractConnectorSubtask:118 - PipeConnectionException occurred, org.apache.iotdb.db.pipe.connector.protocol.thrift.async.IoTDBDataRegionAsyncConnector retries to handshake with the target system. 

[sonarqubecloud]

Quality Gate passed

Issues
0 New issues
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarCloud