Feature/kafka connector

Dockerfile

@@ -9,9 +9,10 @@ RUN apk add --no-cache bash
 
 # Add forklift server
 WORKDIR /tmp
-ADD server/target/universal/forklift-server-0.32.zip forklift.zip
+ADD server/target/universal/forklift-server-0.33.zip forklift.zip
 RUN yes | unzip -d /usr/local forklift.zip
-RUN ln -s /usr/local/forklift-server-0.32 /usr/local/forklift
+RUN ln -s /usr/local/forklift-server-0.33 /usr/local/forklift
+
 RUN rm forklift.zip
 RUN mkdir -p /usr/local/forklift/consumers
 

README.adoc

@@ -8,6 +8,9 @@ full documentation.
 
 link:doc/forklift.adoc[Documentation]
 = Releases
+* 0.33
+** Added support for embedding Forklift Server and runtime deployments
+
 * 0.32
 ** Improve performance of replay indexing.
 

connectors/kafka/.gitignore

@@ -0,0 +1 @@
+/bin/

connectors/kafka/README.md

@@ -0,0 +1,21 @@
+# README #
+
+## Kafka Connector ##
+The kafka connector allows the forklift to utilize kafka and confluent's schema registry.  
+
+## JMS ##
+Forklift was developed against the JMS spec.  Kafka is not a JMS implementation so some adaptation
+has occurred in order to allow Kafka to function with Forklift.
+
+### Queues and Topics ###
+Kafka only provides one form of topic which is analogous to ActiveMQ's Virtual Topic.  Only one member from
+each subscribed group will receive a message.  Requests for a queue or topic will return the 
+same producer type.
+
+### At Most Once Delivery ###
+The kafka connector does not guarantee at most once delivery although it makes a best effort. 
+In order to adapt to the JMS spec, messages are acknowledged and added to a pending commit batch, 
+which are committed to kafka every poll cycle.  It is therefore possible for the server to crash 
+before the acknowledgment batch has been committed, and after the message has processed.
+
+

connectors/kafka/build.sbt

@@ -0,0 +1,82 @@
+organization := "com.github.dcshock"
+
+name := "forklift-kafka"
+
+version := "0.1"
+
+// target and Xlint cause sbt dist to fail
+javacOptions ++= Seq("-source", "1.8")//, "-target", "1.8", "-Xlint")
+
+javacOptions in compile ++= Seq("-g:lines,vars,source")
+
+initialize := {
+  val _ = initialize.value
+  if (sys.props("java.specification.version") != "1.8")
+    sys.error("Java 8 is required for this project.")
+}
+
+libraryDependencies ++= Seq(
+  "com.github.dcshock" % "forklift" % "0.19" ,
+  "com.fasterxml.jackson.core" % "jackson-databind" % "2.7.3",
+  "com.fasterxml.jackson.datatype" % "jackson-datatype-jsr310" % "2.7.3",
+  "org.apache.kafka" % "kafka-clients" % "0.10.1.1-cp1",
+  "io.confluent" % "kafka-avro-serializer" % "3.1.1"
+
+)
+
+lazy val testDependencies = Seq(
+  "commons-io" % "commons-io" % "2.4" ,
+  "com.novocode" % "junit-interface" % "0.11",
+  "org.mockito"       % "mockito-core"            % "1.9.5"
+)
+
+libraryDependencies ++= testDependencies.map(_ % "test")
+
+resolvers ++= Seq(
+  "Sonatype OSS Snapshots" at "https://oss.sonatype.org/content/repositories/snapshots",
+  "Maven Central" at "http://repo1.maven.org/maven2",
+  "Fuse Snapshots" at "http://repo.fusesource.com/nexus/content/repositories/snapshots",
+  "Fuse" at "http://repo.fusesource.com/nexus/content/groups/public",
+  "Confluent Maven Repo" at "http://packages.confluent.io/maven/"
+)
+
+// Remove scala dependency for pure Java libraries
+autoScalaLibrary := false
+
+// Remove the scala version from the generated/published artifact
+crossPaths := false
+
+publishMavenStyle := true
+
+publishTo := {
+  val nexus = "https://oss.sonatype.org/"
+  if (isSnapshot.value)
+    Some("snapshots" at nexus + "content/repositories/snapshots")
+  else
+    Some("releases"  at nexus + "service/local/staging/deploy/maven2")
+}
+
+pomIncludeRepository := { _ => false }
+
+pomExtra := (
+  <url>https://github.com/dcshock/forklift</url>
+    <licenses>
+      <license>
+        <name>BSD-style</name>
+        <url>http://www.opensource.org/licenses/bsd-license.php</url>
+        <distribution>repo</distribution>
+      </license>
+    </licenses>
+    <scm>
+      <url>git@github.com:dcshock/forklift.git</url>
+      <connection>scm:git:git@github.com:dcshock/forklift.git</connection>
+    </scm>
+    <developers>
+      <developer>
+        <id>dcshock</id>
+        <name>Matt Conroy</name>
+        <url>http://www.mattconroy.com</url>
+      </developer>
+    </developers>)
+
+useGpg := true

connectors/kafka/project/Dependencies.scala

@@ -0,0 +1,5 @@
+import sbt._
+
+object Dependencies {
+  lazy val scalaTest = "org.scalatest" %% "scalatest" % "3.0.1"
+}

connectors/kafka/project/build.properties

@@ -0,0 +1 @@
+sbt.version=0.13.13

connectors/kafka/project/plugins.sbt

@@ -0,0 +1 @@
+addSbtPlugin("com.jsuereth" % "sbt-pgp" % "1.0.0")

connectors/kafka/src/main/java/forklift/connectors/AcknowledgedRecordHandler.java

@@ -0,0 +1,146 @@
+package forklift.connectors;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.common.TopicPartition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Maintains a batch of acknowledged records and provides management for adding and removing {@link org.apache.kafka.common.TopicPartition partitions}.
+ * Acknowledged records are those that have started processing but have not yet been committed to the Kafka Broker.  This class is threadsafe.
+ */
+public class AcknowledgedRecordHandler {
+    private static final Logger log = LoggerFactory.getLogger(AcknowledgedRecordHandler.class);
+    private Map<TopicPartition, OffsetAndMetadata> pendingOffsets = new ConcurrentHashMap<>();
+    private Object pausedLock = new Object();
+    private Object unpausedLock = new Object();
+    private AtomicInteger acknowledgeEntryCount = new AtomicInteger(0);
+    private volatile boolean acknowledgementsPaused = false;
+    private Set<TopicPartition> assignment = ConcurrentHashMap.newKeySet();
+
+    /**
+     * Acknowledges that a record has been received before processing begins.  True is returned if processing should occur else false.
+     * Only records belonging to {@link #addPartitions(java.util.Collection) added partitions} may be processed. Note that this is a
+     * blocking method and a short delay may occur should the available topic paritions be changing.
+     *
+     * @param record
+     * @return true if the record has been achnowledged and may be processed, else false
+     * @throws InterruptedException
+     */
+    public boolean acknowledgeRecord(ConsumerRecord<?, ?> record) throws InterruptedException {
+        boolean acknowledged = false;
+        synchronized (unpausedLock) {
+            while (acknowledgementsPaused) {
+                unpausedLock.wait();
+            }
+            acknowledgeEntryCount.incrementAndGet();
+        }
+        if (!assignment.contains(new TopicPartition(record.topic(), record.partition()))) {
+            acknowledged = false;
+        } else {
+            long offset = record.offset() + 1;
+            TopicPartition topicPartition = new TopicPartition(record.topic(), record.partition());
+            synchronized(this) {
+                if (!pendingOffsets.containsKey(topicPartition) || pendingOffsets.get(topicPartition).offset() < offset) {
+                    OffsetAndMetadata offsetAndMetadata = new OffsetAndMetadata(offset, "Commit From Forklift Server");
+                    pendingOffsets.put(topicPartition, offsetAndMetadata);
+                }
+            }
+            acknowledged = true;
+        }
+
+        synchronized (pausedLock) {
+            int count = acknowledgeEntryCount.decrementAndGet();
+            if (acknowledgementsPaused && count == 0) {
+                pausedLock.notifyAll();
+            }
+        }
+        return acknowledged;
+    }
+
+    /**
+     * Removes and returns the highest offsets of any acknowledged records.
+     * This is a blocking method as a short delay may occur while any threads which are currently acknowledging records are allowed to
+     * complete and any incoming threads are paused.
+     *
+     * @return a Map of the highest offset data for any acknowledged records
+     * @throws InterruptedException if interrupted
+     */
+    public Map<TopicPartition, OffsetAndMetadata> flushAcknowledged() throws InterruptedException {
+        try {
+            this.pauseAcknowledgments();
+            Map<TopicPartition, OffsetAndMetadata> flushed = pendingOffsets;
+            pendingOffsets = new ConcurrentHashMap<>();
+            return flushed;
+        } catch (InterruptedException interrupt) {
+            Thread.currentThread().interrupt();
+            throw interrupt;
+        } catch (Throwable e) {
+            log.error("Error flushing Acknowledged", e);
+            throw e;
+        } finally {
+            this.unpauseAcknowledgements();
+        }
+    }
+
+    /**
+     * Adds additional partitions to be managed.  Only added partitions can be
+     * {@link #acknowledgeRecord(org.apache.kafka.clients.consumer.ConsumerRecord) acknowledged}
+     *
+     * @param addedPartitions the partitions to add
+     */
+    public void addPartitions(Collection<TopicPartition> addedPartitions) {
+        this.assignment.addAll(addedPartitions);
+    }
+
+    /**
+     * Remove partitions from management.  Any existing offsets for the removed partitions are returned.  Note that the offest is the highest
+     * acknowleged message's offset + 1 per kafka's specification of how to commit offsets.  Note that this
+     * is a blocking method as any threads which are currently acknowledging records are allowed to complete and any
+     * incoming threads are paused.
+     *
+     * @param removedPartitions the partitions to remove
+     * @return the highest offsets of the removed partitions
+     * @throws InterruptedException
+     */
+    public Map<TopicPartition, OffsetAndMetadata> removePartitions(Collection<TopicPartition> removedPartitions)
+                    throws InterruptedException {
+        pauseAcknowledgments();
+        try {
+            Map<TopicPartition, OffsetAndMetadata> removedOffsets = new HashMap<>();
+            for (TopicPartition topicPartition : removedPartitions) {
+                if (pendingOffsets.containsKey(topicPartition)) {
+                    removedOffsets.put(topicPartition, pendingOffsets.remove(topicPartition));
+                }
+            }
+            assignment.removeAll(removedPartitions);
+            return removedOffsets;
+        } finally {
+            unpauseAcknowledgements();
+        }
+    }
+
+    private void pauseAcknowledgments() throws InterruptedException {
+        synchronized (pausedLock) {
+            acknowledgementsPaused = true;
+            while (acknowledgeEntryCount.get() != 0) {
+                pausedLock.wait();
+            }
+        }
+    }
+
+    private void unpauseAcknowledgements() {
+        synchronized (unpausedLock){
+            acknowledgementsPaused = false;
+            unpausedLock.notifyAll();
+        }
+    }
+
+}