Restore messages from s3

[MacRudDNA]

…t of records

[loffek]

LGTM! But see my question around de-dup logic

[MacRudDNA]

How it works.
Based on topics to restore allow and deny list we get list of TopicPartitionToRestore objects. For each there will be run one worker in thread pool with its own transactional producer.

Worker obtains list of json files with backup data and tries to send content of each batch in one transaction. After sending each message it's offset is saved in memory to prevent duplicates - before sending message we check if it's offset has already ben produced.
After committing transaction we read metadata of produced records to update new offset that each of the messages got.
So we end up with map of old-new offset for each restored message.

It is possible to force producer to use transaction for one produced record - to do this one needs to change RestoreMessageProducer.PRODUCER_BATCH_SIZE to 1.

Please note that we produce each record with timestamp we have from backup. Looking at documentation HERE I would argue that we can always give timestamp to produced message but for CreateTime type we can get some messages not being restored if restoring old messages. On the other hand if message.timestamp.difference.max.ms is same as retention.ms it should make no difference for us because older messages would also disappear from source cluster.

[loffek]

👨‍🍳

[loffek]

Please note that we produce each record with timestamp we have from backup. Looking at documentation HERE I would argue that we can always give timestamp to produced message but for CreateTime type we can get some messages not being restored if restoring old messages. On the other hand if message.timestamp.difference.max.ms is same as retention.ms it should make no difference for us because older messages would also disappear from source cluster.

💯