feat: consume data by batch

[Sotatek-HuyLe3a]

#55

[satran004]

Thanks @Sotatek-HuyLe3a . Few comments otherwise looks good to me.

[satran004]

@Sotatek-HuyLe3a I think we can use CursorService.rollbackTo(slot) method here instead of directly accessing CursorStorage.

[satran004]

@Sotatek-HuyLe3a Yaci Store should automatically start this if autoSync is true. Do we need to handle it separately here?

[Sotatek-HuyLe3a]

Thanks @satran004 , My intention is to start the Yaci Store service if syncStart is false, however, my current approach will make the autoSync property meaningless. I will fix it.

[satran004]

@Sotatek-HuyLe3a Few more questions.

1. For regular synchronization (after a full sync), is there any impact during a rollback? Or does it not matter because the batch size is changed to 1 during regular sync? Should we explicitly set the batch size to 1 when eventMetadata.isSync() becomes true?

2. What about rollback during a restart? Will it be handled properly during batch processing? This is because, during a restart, yaci-store sends a rollback event for the last block.

[Sotatek-HuyLe3a]

@Sotatek-HuyLe3a Few more questions.

1. For regular synchronization (after a full sync), is there any impact during a rollback? Or does it not matter because the batch size is changed to 1 during regular sync? Should we explicitly set the batch size to 1 when eventMetadata.isSync() becomes true?
2. What about rollback during a restart? Will it be handled properly during batch processing? This is because, during a restart, yaci-store sends a rollback event for the last block.

1. With the current logic, we do not always consume based on a fixed batch size. When crawling to the tip, we will consume one by one because we have condition (BlockEventListener.java): "lastReceivedTimeElapsed >= commitThreshold."
   if (currentBlockCount % batchSize == 0 || lastReceivedTimeElapsed >= commitThreshold) { blockSyncService.startBlockSyncing(); blockCount.set(0); }

However, since we can start BlockSync when we are close to the tip, and I think there's no need to set the batch size value to 1, we can add one more condition:
if (currentBlockCount % batchSize == 0 || lastReceivedTimeElapsed >= commitThreshold || eventMetadata.isSyncMode()) { blockSyncService.startBlockSyncing(); blockCount.set(0); }

2.If we process a batch and encounter a failure, the data will be rolled back, except for the data in the cursor table, and we handled this issue when ledger-sync restart. Each time there is a restart, Yaci Store publishes a rollback event to the last block as a way to clean the data and I think this does not have any impact on batch processing.

[satran004]

Thanks @Sotatek-HuyLe3a

if (currentBlockCount % batchSize == 0 || lastReceivedTimeElapsed >= commitThreshold || eventMetadata.isSyncMode()) { blockSyncService.startBlockSyncing(); blockCount.set(0); }

Makes sense. Can you please change the code accordingly? Then we can merge this PR. Thanks