<commit_before># fs2-kafka
Kafka client implemented with [fs2](https://github.com/functional-streams-for-scala/fs2) library

[![Build Status](https://travis-ci.org/Spinoco/fs2-kafka.svg?branch=master)](https://travis-ci.org/Spinoco/fs2-kafka)
[![Gitter Chat](https://badges.gitter.im/functional-streams-for-scala/fs2.svg)](https://gitter.im/fs2-http/Lobby)
 
## Overview 

fs2-kafka is a simple client for consuming and publishing messages from / to Apache Kafka. It has minimalistic functionality and is fully implemented without any dependency on kafka native driver. It is fully asynchronous and non-blocking. 

Thanks to fs2, it comes with full backpressure and streaming support out of the box, and plays well with other fs2 libraries such as [fs2-http](https://github.com/Spinoco/fs2-http). 

fs2-kafka was built with minimal dependencies (apart from fs2, only scodec and shapeless is used). 

fs2-kafka only supports a subset of features compared to the native kafka client. There is for example no support for creating/administering topics and partitions, or for Kafka Connect / Kafka Streams. The reason for that is that we wanted the client to be as simple as possible, thus leaving all non-essential functionality to other solutions. Please note there is also no support for SASL Kafka protocol now. 



## Features

- Subscribe to topic / parition, with configurable pre-fetch behaviour.
- Publish to topic / partition 
- Query metadata from kafka cluster
- Query offset range for topic / partition 

For the compression of messages fs2-kafka supports GZIP and Snappy compression. 

fs2-kafka supports versions between 0.8.2 and 0.10.2 of kafka cluster with respective protocols. The protocols are cross-tested against the different versions of Kafka (i.e. 0.10.2 is tested to wrok ok with 0.8.2, 0.9.1, 0.10, 0.10.1 and 0.10.2 protocols). 

## SBT

Add this to your build file

```
libraryDependencies += "com.spinoco" %% "fs2-kafka" % "0.1.2" 
```

## Dependencies

version  |    scala  |   fs2  |  scodec | shapeless      
---------|-----------|--------|---------|-----------
0.1.2    | 2.11, 2.12|  0.9.7 | 1.10.3  | 2.3.2 


## Usage 

Throughout this simple usage guide, please consider having the following imports on your classpath:

```scala
import fs2.util.syntax._
import spinoco.fs2.kafka
import spinoco.fs2.kafka._
import spinoco.protocol.kafka._
import scala.concurrent.duration._
```
If you type console from fs2-kafka's project sbt, these imports are added for you automatically. 

### Basics

The client can be used with any `F` (effectfull type) that conforms to `Async[F]`. As a reference throughout the guide we will use fs2.Task. 

To obtain a Kafka client the following code can be used :

```scala

kafka.client(
  ensemble = Set(broker("kafka-broker1-dns-name", port = 9092))<commit_msg>Updated README to contain new RC version.<commit_after># fs2-kafka
Kafka client implemented with [fs2](https://github.com/functional-streams-for-scala/fs2) library

[![Build Status](https://travis-ci.org/Spinoco/fs2-kafka.svg?branch=master)](https://travis-ci.org/Spinoco/fs2-kafka)
[![Gitter Chat](https://badges.gitter.im/functional-streams-for-scala/fs2.svg)](https://gitter.im/fs2-http/Lobby)
 
## Overview 

fs2-kafka is a simple client for consuming and publishing messages from / to Apache Kafka. It has minimalistic functionality and is fully implemented without any dependency on kafka native driver. It is fully asynchronous and non-blocking. 

Thanks to fs2, it comes with full backpressure and streaming support out of the box, and plays well with other fs2 libraries such as [fs2-http](https://github.com/Spinoco/fs2-http). 

fs2-kafka was built with minimal dependencies (apart from fs2, only scodec and shapeless is used). 

fs2-kafka only supports a subset of features compared to the native kafka client. There is for example no support for creating/administering topics and partitions, or for Kafka Connect / Kafka Streams. The reason for that is that we wanted the client to be as simple as possible, thus leaving all non-essential functionality to other solutions. Please note there is also no support for SASL Kafka protocol now. 



## Features

- Subscribe to topic / parition, with configurable pre-fetch behaviour.
- Publish to topic / partition 
- Query metadata from kafka cluster
- Query offset range for topic / partition 

For the compression of messages fs2-kafka supports GZIP and Snappy compression. 

fs2-kafka supports versions between 0.8.2 and 0.10.2 of kafka cluster with respective protocols. The protocols are cross-tested against the different versions of Kafka (i.e. 0.10.2 is tested to wrok ok with 0.8.2, 0.9.1, 0.10, 0.10.1 and 0.10.2 protocols). 

## SBT

Add this to your build file

```
libraryDependencies += "com.spinoco" %% "fs2-kafka" % "0.1.2" 
```

## Dependencies

version  |    scala  |   fs2     |  scodec | shapeless      
---------|-----------|-----------|---------|-----------
0.1.2    | 2.11, 2.12|      0.9.7| 1.10.3  | 2.3.2 
0.2.0-RC1| 2.11, 2.12| 0.10.0-RC2| 1.10.3  | 2.3.2


## Usage 

Throughout this simple usage guide, please consider having the following imports on your classpath:

```scala
import fs2.util.syntax._
import spinoco.fs2.kafka
import spinoco.fs2.kafka._
import spinoco.protocol.kafka._
import scala.concurrent.duration._
```
If you type console from fs2-kafka's project sbt, these imports are added for you automatically. 

### Basics

The client can be used with any `F` (effectfull type) that conforms to `Async[F]`. As a reference throughout the guide we will use fs2.Task. 

To obtain a Kafka client the following code can be used :

```scala

kafka.client(
  ensemble = Set(broker("kafka-broker1-dns-name", port = 9092))
{'fraction_non_alphanumeric': 0.11074293686780606, 'fraction_numerical': 0.03139169863969306, 'mean_word_length': 3.577015163607342, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 12, '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': '3449262', 'n_tokens_mistral': 1888, 'n_tokens_neox': 1734, 'n_words': 677}