Streaming composition Traversal, data cannot be inserted

**Describe the bug**
A clear and concise description of what the bug is.
Streaming composition Traversal, data cannot be inserted,As a result, the cyclic trading results cannot be reproduced.
**Expected behavior**
A clear and concise description of what you expected to happen.
2025-08-01 14:17:51,517 [shuffle-reader-7[0/2]] INFO PipelineInputFetcher:211 - fetcher task finish window id 884 shuffle-reader-7[0/2]
2025-08-01 14:17:51,879 [stats-flusher-0] INFO JdbcKVStore:74 - key: geaflow1753927737412687237_metrics_CYCLE_PipelineTask#1-884_cycle#1[1], insert fail, try insert
2025-08-01 14:17:51,883 [stats-flusher-0] INFO JdbcKVStore:74 - key: geaflow1753927737412687237_metrics_CYCLE_PipelineTask#1-884_cycle#1[8], insert fail, try insert
2025-08-01 14:17:51,886 [stats-flusher-0] INFO JdbcKVStore:74 - key: geaflow1753927737412687237_metrics_CYCLE_PipelineTask#1-884_cycle#1[9], insert fail, try insert
2025-08-01 14:17:51,890 [stats-flusher-0] INFO JdbcKVStore:74 - key: geaflow1753927737412687237_metrics_CYCLE_PipelineTask#1-884_cycle#1[13], insert fail, try insert
2025-08-01 14:17:51,894 [stats-flusher-0] INFO JdbcKVStore:74 - key: geaflow1753927737412687237_metrics_CYCLE_PipelineTask#1-884_cycle#1[14], insert fail, try insert
2025-08-01 14:17:51,897 [stats-flusher-0] INFO JdbcKVStore:74 - key: geaflow1753927737412687237_metrics_CYCLE_PipelineTask#1-884_cycle#2-1[15], insert fail, try insert
2025-08-01 14:17:51,901 [stats-flusher-0] INFO JdbcKVStore:74 - key: geaflow1753927737412687237_metrics_CYCLE_PipelineTask#1-884_cycle#2-2[15], insert fail, try insert
2025-08-01 14:17:51,905 [stats-flusher-0] INFO JdbcKVStore:74 - key: geaflow1753927737412687237_metrics_CYCLE_PipelineTask#1-884_cycle#2-3[15], insert fail, try insert
2025-08-01 14:17:51,909 [stats-flusher-0] INFO JdbcKVStore:74 - key: geaflow1753927737412687237_metrics_CYCLE_PipelineTask#1-884_cycle#2-4[15], insert fail, try insert
2025-08-01 14:17:51,913 [stats-flusher-0] INFO JdbcKVStore:74 - key: geaflow1753927737412687237_metrics_CYCLE_PipelineTask#1-884_cycle#2-5[15], insert fail, try insert
2025-08-01 14:17:51,917 [stats-flusher-0] INFO JdbcKVStore:74 - key: geaflow1753927737412687237_metrics_CYCLE_PipelineTask#1-884_cycle#3[18], insert fail, try insert
2025-08-01 14:17:51,924 [stats-flusher-0] INFO JdbcKVStore:74 - key: geaflow1753927737412687237_metrics_PIPELINE_PipelineTask#1-885, insert fail, try insert
2025-08-01 14:17:51,928 [stats-flusher-0] INFO BaseJdbcStore:148 - flush
2025-08-01 14:17:52,931 [stats-flusher-0] INFO BaseJdbcStore:148 - flush
2025-08-01 14:18:21,431 [geaflow-worker-2] INFO GeaFlowTableSourceFunction:202 - fetch data size: 0, isFinish: false, table: table_new_trade, partition: tbl-trade-0, batchId: 885,nextOffset: 2025-08-01 14:18:21, cost 30008
2025-08-01 14:18:21,431 [geaflow-worker-2] INFO KafkaConsumer:1123 - [Consumer clientId=consumer-default-group-id-3, groupId=default-group-id] Subscribed to partition(s): tbl-trade-2
2025-08-01 14:18:21,431 [geaflow-worker-2] INFO KafkaConsumer:1564 - [Consumer clientId=consumer-default-group-id-3, groupId=default-group-id] Seeking to offset 0 for partition tbl-trade-2
2025-08-01 14:18:22,940 [stats-flusher-0] INFO BaseJdbcStore:148 - flush
2025-08-01 14:18:51,439 [geaflow-worker-2] INFO GeaFlowTableSourceFunction:202 - fetch data size: 0, isFinish: false, table: table_new_trade, partition: tbl-trade-2, batchId: 885,nextOffset: 2025-08-01 14:18:51, cost 30008
2025-08-01 14:18:51,439 [geaflow-worker-2] INFO KafkaConsumer:1123 - [Consumer clientId=consumer-default-group-id-3, groupId=default-group-id] Subscribed to partition(s): tbl-trade-1
2025-08-01 14:18:51,439 [geaflow-worker-2] INFO KafkaConsumer:1564 - [Consumer clientId=consumer-default-group-id-3, groupId=default-group-id] Seeking to offset 0 for partition tbl-trade-1
2025-08-01 14:18:52,946 [stats-flusher-0] INFO BaseJdbcStore:148 - flush
**Additional context**
Add any other context about the problem here.
https://tugraph-analytics.github.io/opinion/2023/06/16/%E8%B0%81%E5%9C%A8%E4%BB%A5%E5%A4%AA%E5%9D%8A%E5%8C%BA%E5%9D%97%E9%93%BE%E4%B8%8A%E5%BE%AA%E7%8E%AF%E4%BA%A4%E6%98%93-TuGraph+Kafka%E7%9A%840%E5%85%83%E6%B5%81%E5%9B%BE%E8%A7%A3%E5%86%B3%E6%96%B9%E6%A1%88.html

Start by tracing the logged failure through PipelineInputFetcher, JdbcKVStore, and BaseJdbcStore, then inspect GeaFlowTableSourceFunction and KafkaConsumer around the repeated partition reads. Reproduce the streaming composition Traversal with the cyclic trading workload and verify that data is inserted and the cyclic trading results can be reproduced.

Written by the indexing model from the issue text.