Tracking issue for remote WAL based on Kafka

[MichaelScofield]

What type of enhancement is this?

Feature

What does the enhancement do?

Tracking tasks:

• Skeleton:
  • Add skeleton codes for remote wal - meta srv. feat(remote_wal): add skeleton for remote wal related to meta srv #2933
  • Add skeleton codes for remote wal - datanode. feat(remote_wal): add skeleton for remote wal related to datanode #2941
• Metasrv
  • Allocate wal options upon handling create table requests. feat(remote_wal): implement topic allocation #2970
• Datanode
  • Implement kafka log store. feat(remote_wal): impl kafka log store #2971
  • Integrate kafka log store into the wal system. refactor(remote_wal): entry id usage #2986
• Tests
  • Integration test.
    • Setup a kafka cluster for CI. chore: setup kafka standalone in coverage test #2984
    • Add integration tests for remote wal. test(remote_wal): add unit tests for kafka remote wal #2993

Implementation challenges

No response

[tisonkun]

Roughtly share some thoughts:

1. For the client library, you may prefer rdkafka which is endorsed by Confluent and features in sync with the maintained librdkafka.
2. For the log store API, with a prototype I find the following topics to be determined:

First, Entry's ID should be generalized if we keep this interface:

/// Create a new `EntryStream` to asynchronously generates `Entry` with ids
    /// starting from `id`.
    async fn read(
        &self,
        ns: &Self::Namespace,
        id: Id,
    ) -> Result<SendableEntryStream<Self::Entry, Self::Error>, Self::Error>;

Cause Kafka's seek takes both partition and offset (at least two integeters).

Second, we should decide how to map Namespace to Kafka's topic. Simply, config a base topic for the log store (wal_dir?), and create a new topic for each namespace (format!("{wal_dir}-{ns}")).

The other topics would be how much options we should include in the first version, and if we just let message retention to GC.

BTW, this doesn't seems a "Tech debt reduction" but a new feature?

[tisonkun]

With rdkafka you can simply implement append with FutureProducer like:

        let topic = format!("BASE-TOPIC-{}", e.ns);
        let (partition, offset) = self
            .producer
            .send(
                FutureRecord::to(&topic).payload(&e.data).key(&e.id),
                Duration::from_secs(0),
            )
            .await?;
        Ok(todo_partition_offset_to_id(partition, offset))

and read with StreamConsumer, namespace manipulation with AdminClient<DefaultClientContext>.

[MichaelScofield]

@tisonkun thx for your suggestions! rdkafka is looking great. However, for the usage of kafka as a remote wal service, we prefer rskafka. See the reasons here.

As to the kafka topic and partition design, we are taking a very special way. Will illustrate it in the following RFC, stay tune if you are interested!

It's kind of "tech debt reduction" because it's a long awaiting feature that's planned since early days.

[tisonkun]

However, for the usage of kafka as a remote wal service, we prefer rskafka. See the reasons here.

This makes sense. Good to know :D

[tisonkun]

@MichaelScofield I noticed that part of the tracking items are checked. Are they implemented in a personal branch or privately?

[niebayes]

@tisonkun The remote wal project is split into several tasks. Most of them are implemented and are waiting to be merged.
See: #2753

[evenyag]

The remote wal project is split into several tasks. Most of them are implemented and are waiting to be merged. See: #2753

@niebayes Could you please link your pull requests to the tasks? I think we should mark the item as checked after the related pull request is merged otherwise it is confusing. e.g. I can't find KafkaLogStore in our codebase but it is already checked.

[niebayes]

@evenyag Unmarked and linked related issues or pull requests.