feat(remote_wal): impl kafka log store

[niebayes]

I hereby agree to the terms of the GreptimeDB CLA

What's changed and what's your intention?

• Implemented a ClientManager to manage construction and accesses of rskafka client. Through the manager, the kafka log store is able to contact kafka cluster.
• Implemented a KafkaLogStore to provide read/write interfaces for interacting with Kafka remote wal.
• Built and initialized KafkaLogStore upon starting the datanode.

TODO

• Integration tests.

Checklist

• I have written the necessary rustdoc comments.
• I have added the necessary unit tests and integration tests.

Refer to a related PR or issue link (optional)

[codecov]

Codecov Report

Attention: 186 lines in your changes are missing coverage. Please review.

Comparison is base (d4ac873) 85.83% compared to head (0f4ef3e) 85.72%.

Additional details and impacted files

@@             Coverage Diff             @@
##           develop    #2971      +/-   ##
===========================================
- Coverage    85.83%   85.72%   -0.11%     
===========================================
  Files          780      783       +3     
  Lines       126196   126435     +239     
===========================================
+ Hits        108319   108389      +70     
- Misses       17877    18046     +169     

[v0y4g3r]

Had an offline discussion with @WenyXu . We all agree that the split/merge of WAL entries should happen inside logstore. It's the logstore that know the exact log entry size limit and how to handle the split/merge. AFAIK rskafka employs some batch mechanism so that we can simply forwrd wal entries to Kafka records in a 1:1 manner and let rskafka to handle the batching stuff. Only one thing to notice: be careful to handle large single wal entries, split those large entries inside logstore and ensure the splited parts get produced sequentially and atomically.

[niebayes]

Splitting a way-too large log entry is in planning. However, it may be implemented in a subsequent PR since we need to investigate how rskafka client handles a way-too large log entry and then design an appropriate splitting strategy.

[killme2008]

Almost LGTM

[fengjiachun]

LGTM