[KAFKA-13239] Use RocksDB.ingestExternalFile for restoration

• Type: Improvement
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Component/s: streams
• Assignee: Unassigned
• Reporter: Guozhang Wang
• Created: 27/Aug/21 06:58
• Updated: 22/Sep/21 16:40

Description

Now that we are in newer version of RocksDB, we can consider using the new

ingestExternalFile(final ColumnFamilyHandle columnFamilyHandle,
      final List<String> filePathList,
      final IngestExternalFileOptions ingestExternalFileOptions)

for restoring changelog into state stores. More specifically:

1) Use larger default batch size in restore consumer polling behavior so that each poll would return more records as possible.
2) For a single batch of records returned from a restore consumer poll call, first write them as a single SST File using the SstFileWriter. The existing DBOptions could be used to construct the EnvOptions} and {{Options for the writter.
Do not yet ingest the written file to the db yet within each iteration
3) At the end of the restoration, call RocksDB.ingestExternalFile given all the written files' path as the parameter. The IngestExternalFileOptions would be specifically configured to allow key range overlapping with mem-table.
4) A specific note is that after the call in 3), heavy compaction may be executed by RocksDB in the background and before it cools down, starting normal processing immediately which would try to put new records into the store may see high stalls. To work around it we would consider using RocksDB.compactRange() which would block until the compaction is completed.

Issue Links

• is related to: KAFKA-8627 Investigate batching on state restore
• relates to: KAFKA-10199 Separate state restoration into separate threads