<commit_before>finagle-kafka
=============

An Apache Kafka client in Netty and Finagle.

## Overview
*finagle-kafka* is an [Apache Kafka](https://kafka.apache.org/) client in [Netty](http://netty.io/)
and [Twitter's Finagle](http://twitter.github.io/finagle/). It enables you to handle Kafka in more
functional and composable way based on [Futures](http://twitter.github.io/finagle/guide/Futures.html).

You can also utilize excellent features of Finagle: *load balancing, connection pooling, timeouts,
retries and bunch of statistics for monitoring and diagnostics.*

**Your feedbacks and contributions are welcome!**

## Build
```
$ git clone https://github.com/okapies/finagle-kafka.git
$ cd finagle-kafka
$ sbt package
```

Hosting on maven repository is not ready at this time (available within days).

## Usage
```
import com.twitter.util.Future
import okapies.finagle._
import okapies.finagle.kafka.protocol._

// connect to bootstrap broker
val bootstrap = Kafka.newRichClient("[host]:[port]")

// create a client for the leader of specific topic partition
val client = for {
  metadata <- bootstrap.metadata("topic")
  leader <- Future.value(metadata.head.partitions(0).leader.get)
  client <- Future.value(Kafka.newRichClient(s"${leader.host}:${leader.port}"))
} yield client

// get offset
val offset = client.map(_.offset("topic", 0, -1).get.offsets(0)).get

// produce a message
client.foreach(_.produce("topic", 0, "foobar"))

// fetch messages
val msgs = client.flatMap {
  _.fetch("topic", 0, offset).map {
    _.messages.map(_.message.value.toString("UTF-8"))
  }
}
msgs.foreach(_.foreach(println))
```

## Future work
- Compression support
- Partitioning and Zookeeper support
- Migration to Netty 4/5 and Finagle 7
- More high-level and sophisticated `KafkaClient`
- Compatibility test<commit_msg>Added description in the README on how to run tests<commit_after>finagle-kafka
=============

An Apache Kafka client in Netty and Finagle.

## Overview
*finagle-kafka* is an [Apache Kafka](https://kafka.apache.org/) client in [Netty](http://netty.io/)
and [Twitter's Finagle](http://twitter.github.io/finagle/). It enables you to handle Kafka in more
functional and composable way based on [Futures](http://twitter.github.io/finagle/guide/Futures.html).

You can also utilize excellent features of Finagle: *load balancing, connection pooling, timeouts,
retries and bunch of statistics for monitoring and diagnostics.*

**Your feedbacks and contributions are welcome!**

## Build
```
$ git clone https://github.com/okapies/finagle-kafka.git
$ cd finagle-kafka
$ sbt package
```

Hosting on maven repository is not ready at this time (available within days).

## Usage
```
import com.twitter.util.Future
import okapies.finagle._
import okapies.finagle.kafka.protocol._

// connect to bootstrap broker
val bootstrap = Kafka.newRichClient("[host]:[port]")

// create a client for the leader of specific topic partition
val client = for {
  metadata <- bootstrap.metadata("topic")
  leader <- Future.value(metadata.head.partitions(0).leader.get)
  client <- Future.value(Kafka.newRichClient(s"${leader.host}:${leader.port}"))
} yield client

// get offset
val offset = client.map(_.offset("topic", 0, -1).get.offsets(0)).get

// produce a message
client.foreach(_.produce("topic", 0, "foobar"))

// fetch messages
val msgs = client.flatMap {
  _.fetch("topic", 0, offset).map {
    _.messages.map(_.message.value.toString("UTF-8"))
  }
}
msgs.foreach(_.foreach(println))
```

## Running tests

Requires a running kafka and zookeeper on default ports and to add the following
to the kafka config (config/server.properties):

```
delete.topic.enable=true
```

Tests are run using sbt.

```
sbt test
```

## Future work
- Compression support
- Partitioning and Zookeeper support
- Migration to Netty 4/5 and Finagle 7
- More high-level and sophisticated `KafkaClient`
- Compatibility test
{'fraction_non_alphanumeric': 0.13622370446382762, 'fraction_numerical': 0.00513083632632119, 'mean_word_length': 4.57, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 3, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4240804', 'n_tokens_mistral': 1252, 'n_tokens_neox': 1187, 'n_words': 367}