AbsaOSS / AbsaOSS/hyperdrive-trigger

Asynchronous kafka offset requests

オープン
#797 コメント 2 件 リアクション 0 件 担当者 1 名 @filiphornak が担当を希望しています GitHub で見る
backend enhancement refactoring
主要言語
TypeScript
スター
8
フォーク
5
PR マージ指標
30日以内にマージされた PR はありません

説明

## Background
We use kafka consumers to obtain the beginning and end offsets.
Such consumers are not thread-safe and can not be used in multithreaded environments like inside Futures in scala without any consequences.
Creating a new short-lived kafka consumer for single offset requests would impose significant overhead. So, for now, we are using the LRU cache provided by the spring framework to reduce such overhead.
However, because of the lack of thread safety of kafka consumers, newly created consumers must be bound to a single thread and are not reusable by others. This leads to writing blocking code that happens to run in a separate thread.

And even then, can we prevent race conditions? Then again, getting offsets might be resistant to these race conditions. It would need further research.

## Feature
Kafka service should provide an asynchronous interface for getting offsets for a particular topic.

## Example
```scala
trait KafkaService {
def getBeginningOffsets(topic: String, consumerProperties: Properties): Future[Map[Int, Long]]
def getEndOffset(topic: String, consumerProperties: Properties): Future[Map[Int, Long]]
def getBeginningEndOffsets(topic: String, consumerProperties: Properties): Future[BeginningEndOffsets]
}

class KafkaServiceImpl @Inject() (...) extends KafkaService {
private val consumerPool: KafkaConsumerPool = createConsumerPool(config.consumerPoolConfig)

override def getBeginningOffset(topic: String, consumerProperties: Properties): Future[Map[Int, Long]] =
for {
consumer <- consumerPool.acquire(consumerProperties)
offsets <- beginningOffsetOf(consumer, topic)
_ <- consumerPool.release(consumer)
} yield offsets

private def beginningOffsetOf(topic: String)(consumer: Consumer[Any, Any]) =
for {
partitions <- asyncParitionsFor(topic)(consumer)
offsets <- asyncBeginningOffsetsFor(partitions)(consumer)
} yield offsets

private def asyncParititonsFor(topic: String)(consumer: Consumer[Any, Any]): Future[Seq[TopicPartition]] = ???

private def asyncBeginningOffsetFor(partitions: Seq[TopicPartition])(consumer: Consumer[Any, Any]): Future[Map[Int, Long]] = ???
}

class KafkaConsumerPool(...) {
def acquire(properties: Properties): Future[Consumer[Any, Any]] = ???

def release(consumer: Consumer[Any, Any]): Future[Unit] = ???
}
```

## Proposed Solution
Solution Ideas:
1. We can find a client library that does not rely on the kafka consumer and can be shared between threads for requesting topic offsets of a particular kafka cluster. One such library might be [https://docs.spring.io/spring-kafka/reference/html/](https://docs.spring.io/spring-kafka/reference/html/)
2. Research if getting offsets via kafka consumer is susceptible to race conditions.
3. Implement consumer pool as discussed in [https://github.com/AbsaOSS/hyperdrive-trigger/pull/791#discussion_r1129539022](https://github.com/AbsaOSS/hyperdrive-trigger/pull/791#discussion_r1129539022) with @jozefbakus. Here is an article on implementing such a pool [https://medium.com/@sutanu.dalui/kafka-consumer-pooling-cc8e9dfc4d24](https://medium.com/@sutanu.dalui/kafka-consumer-pooling-cc8e9dfc4d24). However, it would still need adaptation to our needs.

コントリビューションガイド

コントリビューションガイドを開く

評価

この issue はまだ評価されていません。

新しい issue をメールで受け取る

初心者向けの GitHub issue を短くまとめたダイジェスト。