Kafka 0.10

.gitignore

@@ -6,3 +6,8 @@ packages
 .packages
 notes.md
 coverage/
+.atom/
+.vscode/
+.idea/
+package_config.json
+.DS_Store

.travis.yml

@@ -1,7 +1,3 @@
-sudo: required
-
-dist: trusty
-
 language: dart
 
 services:
@@ -11,20 +7,15 @@ dart:
   - stable
 
 before_install:
-  - docker build -t kafka-cluster tool/kafka-cluster/
-  - docker run -d --name kafka-cluster -p 2181:2181 -p 9092:9092 -p 9093:9093 --env ADVERTISED_HOST=127.0.0.1 kafka-cluster
-  - docker ps -a
-  - sleep 5
-  - docker exec kafka-cluster bash -c '$KAFKA_HOME/bin/kafka-topics.sh --create --zookeeper=localhost:2181 --topic dartKafkaTest --partitions 3 --replication-factor 2'
-  - docker exec kafka-cluster bash -c '$KAFKA_HOME/bin/kafka-topics.sh --list --zookeeper=localhost:2181'
+  - ./tool/rebuild.sh
 
 script:
-  - pub run test -r expanded test/all.dart
-  - pub global activate coverage
-  - dart --observe=8111 test/all.dart &
-  - sleep 20
-  - pub global run coverage:collect_coverage --port=8111 -o coverage.json --resume-isolates
-  - pub global run coverage:format_coverage --package-root=packages --report-on lib --in coverage.json --out lcov.info --lcov
+  - pub run test test/all.dart
+  # - pub global activate coverage
+  # - dart --observe=8111 test/all.dart &
+  # - sleep 10
+  # - pub global run coverage:collect_coverage --port=8111 -o coverage.json --resume-isolates
+  # - pub global run coverage:format_coverage --packages=.packages --report-on lib --in coverage.json --out lcov.info --lcov
 
 after_success:
-  - bash <(curl -s https://codecov.io/bash)
+  # - bash <(curl -s https://codecov.io/bash)

DEVELOPMENT.md

@@ -4,7 +4,7 @@ Requirements:
 
 * Docker Toolbox (OS X)
 
-## Starting Kafka container locally
+## Kafka 0.8.x
 
 ```
 docker build -t kafka-cluster tool/kafka-cluster/
@@ -22,3 +22,28 @@ Now you should be able to run tests with:
 ```
 pub run test -j 1
 ```
+
+## Kafka 0.10.0.0
+
+We're using `ches/kafka` base image so instructions are a bit different.
+
+```
+# Zookeeper is in a separate container now
+docker run -d --name zookeeper --publish 2181:2181 jplock/zookeeper:3.4.6
+
+# Build our image
+docker build -t kafka tool/kafka/
+
+ZK_IP=$(docker inspect --format '{{ .NetworkSettings.IPAddress }}' zookeeper)
+
+# or for fish users
+export ZK_IP=(docker inspect --format '{{ .NetworkSettings.IPAddress }}' zookeeper)
+
+# Start Kafka container
+docker run -d --name kafka --publish 9092:9092 --publish 9093:9093 \
+  --env KAFKA_ADVERTISED_HOST_NAME=127.0.0.1 \
+  --env ZOOKEEPER_IP=$ZK_IP \
+  kafka
+```
+
+Kafka brokers will be available on `127.0.0.1:9092` and `127.0.0.1:9093`.

README.md

@@ -1,198 +1,79 @@
 # Dart Kafka
 
-[![Build Status](https://travis-ci.org/pulyaevskiy/dart-kafka.svg?branch=master)](https://travis-ci.org/pulyaevskiy/dart-kafka)
+[![Build Status](https://travis-ci.org/dart-kafka/kafka.svg?branch=kafka-0.10)](https://travis-ci.org/dart-kafka/kafka)
 [![Coverage](https://codecov.io/gh/pulyaevskiy/dart-kafka/branch/master/graph/badge.svg)](https://codecov.io/gh/pulyaevskiy/dart-kafka)
 [![License](https://img.shields.io/badge/license-BSD--2-blue.svg)](https://raw.githubusercontent.com/pulyaevskiy/dart-kafka/master/LICENSE)
 
 Kafka client library written in Dart.
 
-### Current status
-
-This library is a work-in-progress and has not been used in production yet.
-
 ### Things that are not supported yet.
 
 * Snappy compression.
+* SSL
 
 ## Installation
 
-There is no Pub package yet, but it will be published as soon as APIs are
-stable enough.
-
-For now you can use git dependency in your `pubspec.yaml`:
-
-```yaml
-dependencies:
-  kafka:
-    git: https://github.com/pulyaevskiy/dart-kafka.git
-```
-
-And then import it as usual:
-
-```dart
-import 'package:kafka/kafka.dart';
-```
-
-## Features
+_To be updated with first release._
 
-This library provides several high-level API objects to interact with Kafka:
-
-* __KafkaSession__ - responsible for managing connections to Kafka brokers and
-  coordinating all requests. Also provides access to metadata information.
-* __Producer__ - publishes messages to Kafka topics
-* __Consumer__ - consumes messages from Kafka topics and stores it's state (current
-  offsets). Leverages ConsumerMetadata API via ConsumerGroup.
-* __Fetcher__ - consumes messages from Kafka without storing state.
-* __OffsetMaster__ - provides convenience on top of Offset API allowing to easily
-  retrieve earliest and latest offsets of particular topic-partitions.
-* __ConsumerGroup__ - provides convenience on top of Consumer Metadata API to easily
-  fetch or commit consumer offsets.
 
 ## Producer
 
-Simple implementation of Kafka producer. Supports auto-detection of leaders for
-topic-partitions and creates separate `ProduceRequest`s for each broker.
-Requests are sent in parallel and all responses are aggregated in special
-`ProduceResult` object.
+Producer publishes messages to the Kafka cluster. Here is a simple example
+of using the producer to send `String` records:
 
 ```dart
-// file:produce.dart
-import 'dart:io';
-import 'package:kafka/kafka.dart';
-
-main(List<String> arguments) async {
-  var host = new ContactPoint('127.0.0.1', 9092);
-  var session = new KafkaSession([host]);
-
-  var producer = new Producer(session, 1, 1000);
-  var result = await producer.produce([
-    new ProduceEnvelope('topicName', 0, [new Message('msgForPartition0'.codeUnits)]),
-    new ProduceEnvelope('topicName', 1, [new Message('msgForPartition1'.codeUnits)])
-  ]);
-  print(result.hasErrors);
-  print(result.offsets);
-  session.close(); // make sure to always close the session when the work is done.
-}
-```
-
-Result:
-
-```shell
-$ dart produce.dart
-$ false
-$ {dartKafkaTest: {0: 213075, 1: 201680}}
-```
-
-## Consumer
-
-High-level implementation of Kafka consumer which stores it's state using
-Kafka's ConsumerMetadata API.
-
-> If you don't want to keep state of consumed offsets take a look at `Fetcher`
-> which was designed specifically for this use case.
-
-Consumer returns messages as a `Stream`, so all standard stream operations
-should be applicable. However Kafka topics are ordered streams of messages
-with sequential offsets. Consumer implementation allows to preserve order of
-messages received from server. For this purpose all messages are wrapped in
-special `MessageEnvelope` object with following methods:
-
-```
-/// Signals to consumer that message has been processed and it's offset can
-/// be committed.
-void commit(String metadata);
-
-/// Signals that message has been processed and we are ready for
-/// the next one. Offset of this message will **not** be committed.
-void ack();
-
-/// Signals to consumer to cancel any further deliveries and close the stream.
-void cancel();
-```
-
-One must call `commit()` or `ack()` for each processed message, otherwise
-Consumer won't send the next message to the stream.
-
-Simplest example of a consumer:
-
-```dart
-import 'dart:io';
 import 'dart:async';
+
 import 'package:kafka/kafka.dart';
 
-void main(List<String> arguments) async {
-  var host = new ContactPoint('127.0.0.1', 9092);
-  var session = new KafkaSession([host]);
-  var group = new ConsumerGroup(session, 'consumerGroupName');
-  var topics = {
-    'topicName': [0, 1] // list of partitions to consume from.
-  };
-
-  var consumer = new Consumer(session, group, topics, 100, 1);
-  await for (MessageEnvelope envelope in consumer.consume(limit: 3)) {
-    // Assuming that messages were produces by Producer from previous example.
-    var value = new String.fromCharCodes(envelope.message.value);
-    print('Got message: ${envelope.offset}, ${value}');
-    envelope.commit('metadata'); // Important.
-  }
-  session.close(); // make sure to always close the session when the work is done.
+Future main() async {
+  var config = new ProducerConfig(bootstrapServers: ['127.0.0.1:9092']);
+  var producer = new Producer<String, String>(
+      new StringSerializer(), new StringSerializer(), config);
+  var record = new ProducerRecord('example-topic', 0, 'key', 'value');
+  producer.add(record);
+  var result = await record.result;
+  print(result);
+  await producer.close();
 }
 ```
 
-It is also possible to consume messages in batches for improved efficiency:
-
-```dart
-import 'dart:io';
-import 'dart:async';
-import 'package:kafka/kafka.dart';
+The producer implements `StreamSink` interface so it is possible to send
+individual records via `add()` as well as streams of records via 
+`addStream()`.
 
-void main(List<String> arguments) async {
-  var host = new ContactPoint('127.0.0.1', 9092);
-  var session = new KafkaSession([host]);
-  var group = new ConsumerGroup(session, 'consumerGroupName');
-  var topics = {
-    'topicName': [0, 1] // list of partitions to consume from.
-  };
-
-  var consumer = new Consumer(session, group, topics, 100, 1);
-  await for (BatchEnvelope batch in consumer.batchConsume(20)) {
-    batch.items.forEach((MessageEnvelope envelope) {
-      // use envelope as usual
-    });
-    batch.commit('metadata'); // use batch control methods instead of individual messages.
-  }
-  session.close(); // make sure to always close the session when the work is done.
-}
-```
+The producer buffers records internally so that they can be sent in bulk to
+the server. This does not necessarily mean increased latency for record
+delivery. When a record is added with `add()` it is sent immediately
+(although an asynchronous gap is present between call to `add()` and actual
+send) as long as there is available slot in IO pool. By default producer
+has a limit of up to 5 in-flight requests per broker connection,
+so delay can occur only if all the slots are already occupied.
 
-### Consumer offset reset strategy
+## Note on new API design
 
-Due to the fact that Kafka topics can be configured to delete old messages
-periodically, it is possible that your consumer offset may become invalid (
-just because there is no such message/offset in Kafka topic anymore).
+Public API of this library has been completely re-written since original 
+version (which supported only Kafka 0.8.x and was never published on Pub).
 
-In such cases `Consumer` provides configurable strategy with following options:
+New design is trying to accomplish two main goals:
 
-* `OffsetOutOfRangeBehavior.throwError`
-* `OffsetOutOfRangeBehavior.resetToEarliest` (default)
-* `OffsetOutOfRangeBehavior.resetToLatest`
+####  1. Follow official Java client semantics and contract.
 
-By default if it gets `OffsetOutOfRange` server error it will reset it's offsets
-to earliest available in the consumed topic and partitions, which essentially
-means consuming all available messages from the beginning.
+`Producer` and `Consumer` are trying to preserve characteristics of
+original Java implementations. This is why configurations for both
+are almost identical to the ones in the official client. The way serialization is implemented also based on Java code.
 
-To modify this behavior simply set `onOffsetOutOfRange` property of consumer to
-one of the above values:
+#### 2. Streams-compatible public API.
 
-```
-var consumer = new Consumer(session, group, topics, 100, 1);
-consumer.onOffsetOutOfRange = OffsetOutOfRangeBehavior.throwError;
-```
+The main reason is to allow better interoperability with other libraries.
+`Producer` implements `StreamSink` for this specific reason, so instead of
+having a `send()` method (as in Java client) there are `add()` and 
+`addStream()`.
 
 ## Supported protocol versions
 
-Current version targets version `0.8.2` of the Kafka protocol. There is no plans
-to support earlier versions.
+Current version targets version `0.10` of the Kafka protocol. 
+There is no plans to support earlier versions.
 
 ## License
 

analysis_options.yaml

@@ -0,0 +1,4 @@
+analyzer:
+  strong-mode:
+    implicit-casts: true
+    implicit-dynamic: true

example/offsets.dart

@@ -0,0 +1,15 @@
+import 'dart:async';
+
+import 'package:kafka/kafka.dart';
+
+Future main() async {
+  var session = Session(['127.0.0.1:9092']);
+  var master = OffsetMaster(session);
+  var offsets = await master.fetchEarliest([
+    TopicPartition('simple_topic', 0),
+    TopicPartition('simple_topic', 1),
+    TopicPartition('simple_topic', 2),
+  ]);
+  print(offsets);
+  await session.close(); // Always close session in the end.
+}

example/simple_consumer.dart

@@ -0,0 +1,25 @@
+import 'dart:async';
+
+import 'package:kafka/kafka.dart';
+import 'package:logging/logging.dart';
+
+Future main() async {
+  Logger.root.level = Level.ALL;
+  Logger.root.onRecord.listen(print);
+
+  var session = Session(['127.0.0.1:9092']);
+  var consumer = Consumer<String, String>(
+      'simple_consumer', StringDeserializer(), StringDeserializer(), session);
+
+  await consumer.subscribe(['simple_topic']);
+  var queue = consumer.poll();
+  while (await queue.moveNext()) {
+    var records = queue.current;
+    for (var record in records.records) {
+      print(
+          "[${record.topic}:${record.partition}], offset: ${record.offset}, ${record.key}, ${record.value}, ts: ${record.timestamp}");
+    }
+    await consumer.commit();
+  }
+  await session.close();
+}

example/simple_producer.dart

@@ -0,0 +1,22 @@
+import 'dart:async';
+import 'package:kafka/kafka.dart';
+
+Future main() async {
+  // Logger.root.level = Level.ALL;
+  // Logger.root.onRecord.listen(print);
+
+  var config = ProducerConfig(bootstrapServers: ['127.0.0.1:9092']);
+  var producer =
+      Producer<String, String>(StringSerializer(), StringSerializer(), config);
+
+  for (var i = 0; i < 10; i++) {
+    // Loop through a list of partitions.
+    for (var p in [0, 1, 2]) {
+      var rec =
+          ProducerRecord('simple_topic', p, 'key:${p},$i', 'value:${p},$i');
+      producer.add(rec);
+      rec.result.then(print);
+    }
+  }
+  await producer.close();
+}