Add fiber implementation for Kafka producer client

build.gradle

@@ -432,6 +432,12 @@ project (':comsat-okhttp') {
     }
 }
 
+project (':comsat-kafka') {
+    dependencies {
+        compile "org.apache.kafka:kafka-clients:0.8.2.2"
+    }
+}
+
 project (':comsat-retrofit') {
     dependencies {
         compile "com.squareup.retrofit:retrofit:$retrofitVer"

comsat-kafka/src/main/java/co/paralleluniverse/fibers/kafka/FiberKafkaProducer.java

@@ -0,0 +1,83 @@
+/*
+ * COMSAT
+ * Copyright (C) 2013-2015, Parallel Universe Software Co. All rights reserved.
+ *
+ * This program and the accompanying materials are dual-licensed under
+ * either the terms of the Eclipse Public License v1.0 as published by
+ * the Eclipse Foundation
+ *
+ *   or (per the licensee's choosing)
+ *
+ * under the terms of the GNU Lesser General Public License version 3.0
+ * as published by the Free Software Foundation.
+ */
+package co.paralleluniverse.fibers.kafka;
+
+import co.paralleluniverse.strands.SettableFuture;
+import org.apache.kafka.clients.producer.*;
+import org.apache.kafka.common.Metric;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.PartitionInfo;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Future;
+
+public class FiberKafkaProducer<K, V> implements Producer<K, V> {
+
+    private final Producer<K, V> producer;
+
+    public FiberKafkaProducer(Producer<K, V> producer) {
+        this.producer = producer;
+    }
+
+    @Override
+    public Future<RecordMetadata> send(ProducerRecord<K, V> record) {
+        return send(record, null);
+    }
+
+    @Override
+    public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
+        SettableFuture<RecordMetadata> future = new SettableFuture<>();
+        producer.send(record, new CallbackWrapper(future, callback));
+        return future;
+    }
+
+    @Override
+    public List<PartitionInfo> partitionsFor(String topic) {
+        return producer.partitionsFor(topic);
+    }
+
+    @Override
+    public Map<MetricName, ? extends Metric> metrics() {
+        return producer.metrics();
+    }
+
+    @Override
+    public void close() {
+        producer.close();
+    }
+
+    private static class CallbackWrapper implements Callback {
+
+        private final SettableFuture<RecordMetadata> future;
+        private final Callback callback;
+
+        public CallbackWrapper(SettableFuture<RecordMetadata> future, Callback callback) {
+            this.future = future;
+            this.callback = callback;
+        }
+
+        @Override
+        public void onCompletion(RecordMetadata metadata, Exception exception) {
+            if (metadata != null) {
+                future.set(metadata);
+            } else {
+                future.setException(exception);
+            }
+            if (callback != null) {
+                callback.onCompletion(metadata, exception);
+            }
+        }
+    }
+}

comsat-kafka/src/test/java/co/paralleluniverse/fibers/kafka/FiberKafkaProducerTest.java

@@ -0,0 +1,208 @@
+/*
+ * COMSAT
+ * Copyright (C) 2013-2015, Parallel Universe Software Co. All rights reserved.
+ *
+ * This program and the accompanying materials are dual-licensed under
+ * either the terms of the Eclipse Public License v1.0 as published by
+ * the Eclipse Foundation
+ *
+ *   or (per the licensee's choosing)
+ *
+ * under the terms of the GNU Lesser General Public License version 3.0
+ * as published by the Free Software Foundation.
+ */
+package co.paralleluniverse.fibers.kafka;
+
+import co.paralleluniverse.fibers.Fiber;
+import co.paralleluniverse.fibers.SuspendExecution;
+import co.paralleluniverse.strands.SuspendableRunnable;
+import org.apache.kafka.clients.producer.Callback;
+import org.apache.kafka.clients.producer.MockProducer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+public class FiberKafkaProducerTest {
+
+    private MockProducer mockProducer;
+    private FiberKafkaProducer<byte[], byte[]> fiberProducer;
+    private CountDownLatch sendCompleteLatch;
+
+    @Before
+    public void setUp() {
+        mockProducer = new MockProducer(false);
+        fiberProducer = new FiberKafkaProducer<>(mockProducer);
+        sendCompleteLatch = new CountDownLatch(1);
+    }
+
+    @Test
+    public void testSuccessfulSendWithoutCallback() throws InterruptedException, TimeoutException, ExecutionException {
+
+        Fiber<Void> fiber = new Fiber<>(new SuspendableRunnable() {
+            @Override
+            public void run() throws SuspendExecution, InterruptedException {
+                Future<RecordMetadata> f = fiberProducer.send(new ProducerRecord<>("Topic", "Key".getBytes(), "Value".getBytes()));
+                Future<RecordMetadata> f2 = fiberProducer.send(new ProducerRecord<>("Topic", "Key".getBytes(), "Value".getBytes()));
+                sendCompleteLatch.countDown();
+                try {
+                    RecordMetadata recordMetadata = f.get();
+
+                    assertEquals("Topic", recordMetadata.topic());
+                    assertEquals(0, recordMetadata.offset());
+                    assertEquals(0, recordMetadata.partition());
+
+                    RecordMetadata recordMetadata2 = f2.get();
+
+                    assertEquals("Topic", recordMetadata2.topic());
+                    assertEquals(1, recordMetadata2.offset());
+                    assertEquals(0, recordMetadata2.partition());
+                } catch (ExecutionException e) {
+                    fail();
+                }
+            }
+        });
+        fiber.start();
+
+        sendCompleteLatch.await(5000, TimeUnit.MILLISECONDS);
+        mockProducer.completeNext();
+        mockProducer.completeNext();
+
+        fiber.join(5000, TimeUnit.MILLISECONDS);
+    }
+
+    @Test
+    public void testSuccessfulSendWithCallback() throws ExecutionException, InterruptedException, TimeoutException {
+        Fiber<Void> fiber = new Fiber<>(new SuspendableRunnable() {
+            @Override
+            public void run() throws SuspendExecution, InterruptedException {
+                final CountDownLatch callbackCompleteLatch = new CountDownLatch(2);
+
+                final AtomicReference<RecordMetadata> callbackMetadata = new AtomicReference<>(null);
+                final AtomicReference<RecordMetadata> callbackMetadata2 = new AtomicReference<>(null);
+
+                Future<RecordMetadata> f = fiberProducer.send(new ProducerRecord<>("Topic", "Key".getBytes(), "Value".getBytes()),
+                        new Callback() {
+                            @Override
+                            public void onCompletion(RecordMetadata metadata, Exception exception) {
+                                callbackMetadata.set(metadata);
+                                callbackCompleteLatch.countDown();
+
+                            }
+                        });
+                Future<RecordMetadata> f2 = fiberProducer.send(new ProducerRecord<>("Topic", "Key".getBytes(), "Value".getBytes()),
+                        new Callback() {
+                            @Override
+                            public void onCompletion(RecordMetadata metadata, Exception exception) {
+                                callbackMetadata2.set(metadata);
+                                callbackCompleteLatch.countDown();
+
+                            }
+                        });
+                sendCompleteLatch.countDown();
+                callbackCompleteLatch.await();
+
+                try {
+                    RecordMetadata recordMetadata = f.get();
+
+                    assertEquals("Topic", recordMetadata.topic());
+                    assertEquals(0, recordMetadata.offset());
+                    assertEquals(0, recordMetadata.partition());
+
+                    assertEquals("Topic", callbackMetadata.get().topic());
+                    assertEquals(0, callbackMetadata.get().offset());
+                    assertEquals(0, callbackMetadata.get().partition());
+
+                    RecordMetadata recordMetadata2 = f2.get();
+
+                    assertEquals("Topic", recordMetadata2.topic());
+                    assertEquals(1, recordMetadata2.offset());
+                    assertEquals(0, recordMetadata2.partition());
+
+                    assertEquals("Topic", callbackMetadata2.get().topic());
+                    assertEquals(1, callbackMetadata2.get().offset());
+                    assertEquals(0, callbackMetadata2.get().partition());
+                } catch (ExecutionException e) {
+                    fail();
+                }
+            }
+        });
+        fiber.start();
+
+        sendCompleteLatch.await(5000, TimeUnit.MILLISECONDS);
+        mockProducer.completeNext();
+        mockProducer.completeNext();
+
+        fiber.join(5000, TimeUnit.MILLISECONDS);
+    }
+
+    @Test
+    public void testErrorSendWithoutCallback() throws ExecutionException, InterruptedException, TimeoutException {
+        final RuntimeException exception = new RuntimeException("Error");
+
+        Fiber<Void> fiber = new Fiber<>(new SuspendableRunnable() {
+            @Override
+            public void run() throws SuspendExecution, InterruptedException {
+                Future<RecordMetadata> f = fiberProducer.send(new ProducerRecord<>("Topic", "Key".getBytes(), "Value".getBytes()));
+                sendCompleteLatch.countDown();
+                try {
+                    f.get();
+                    fail("Send should have failed.");
+                } catch (ExecutionException e) {
+                    assertEquals(exception, e.getCause());
+                }
+            }
+        });
+        fiber.start();
+
+        sendCompleteLatch.await(5000, TimeUnit.MILLISECONDS);
+        mockProducer.errorNext(exception);
+
+        fiber.join(5000, TimeUnit.MILLISECONDS);
+    }
+
+    @Test
+    public void testErrorSendWithCallback() throws ExecutionException, InterruptedException, TimeoutException {
+        final RuntimeException exception = new RuntimeException("Error");
+        final AtomicReference<Exception> exceptionReference = new AtomicReference<>(null);
+
+        Fiber<Void> fiber = new Fiber<>(new SuspendableRunnable() {
+            @Override
+            public void run() throws SuspendExecution, InterruptedException {
+                final CountDownLatch callbackCompleteLatch = new CountDownLatch(1);
+
+                Future<RecordMetadata> f = fiberProducer.send(new ProducerRecord<>("Topic", "Key".getBytes(), "Value".getBytes()),
+                        new Callback() {
+                            @Override
+                            public void onCompletion(RecordMetadata metadata, Exception exception) {
+                                exceptionReference.set(exception);
+                                callbackCompleteLatch.countDown();
+                            }
+                        });
+
+                sendCompleteLatch.countDown();
+                callbackCompleteLatch.await();
+                try {
+                    f.get();
+                    fail("Send should have failed.");
+                } catch (ExecutionException e) {
+                    assertEquals(exception, e.getCause());
+                }
+
+                assertEquals(exception, exceptionReference.get());
+            }
+        });
+        fiber.start();
+
+        sendCompleteLatch.await(5000, TimeUnit.MILLISECONDS);
+        mockProducer.errorNext(exception);
+
+        fiber.join(5000, TimeUnit.MILLISECONDS);
+    }
+}

settings.gradle

@@ -57,3 +57,5 @@ include 'comsat-spring:comsat-spring-boot:comsat-spring-boot-sample-web-secure-j
 include 'comsat-spring:comsat-spring-boot:comsat-spring-boot-sample-web-static'
 include 'comsat-spring:comsat-spring-boot:comsat-spring-boot-sample-web-ui'
 include 'comsat-spring:comsat-spring-boot:comsat-spring-boot-sample-web-velocity'
+include 'comsat-kafka'
+