Kafka Bus Implementation

.gitignore

@@ -1,3 +1,3 @@
 bazel-*
 *~
-.idea/
+.idea/

Gopkg.toml

@@ -86,3 +86,12 @@ required = [
   name = "github.com/knative/serving"
   # HEAD as of 2018-06-29
   revision = "22e3cb0f057daf224ef3af5cc37feb9b152f880e"
+
+[[override]]
+  name = "github.com/Shopify/sarama"
+  revision = "46cf3e2cf1acef7876068f66cf69ec31aad2d0b2" # includes higher level admin client
+#  version = "1.17.0"
+
+[[constraint]]
+  name = "github.com/bsm/sarama-cluster"
+  version = "2.1.13"

config/buses/kafka/README.md

@@ -0,0 +1,33 @@
+# Kafka - Knative Bus
+
+Deployment steps:
+1. Setup [Knative Eventing](../../../DEVELOPMENT.md)
+1. Install a Kafka broker. A simple setup is provided:
+    ```
+    kubectl create namespace kafka
+    kubectl apply -n kafka -f config/buses/kafka/kafka-broker.yaml
+    ```
+1. For cluster wide deployment, change the kind in `config/buses/kafka/kafka-bus.yaml` from `Bus` to `ClusterBus`.
+1. Apply the Kafka Bus:
+    ```
+    ko apply -f config/buses/kafka/kafka-bus.yaml
+    ```
+1. Create Channels that reference the 'kafka' Bus
+1. (Optional) Install [Kail](https://github.com/boz/kail) - Kubernetes tail
+
+The bus has an independent provisioner and dispatcher.
+
+The provisioner will create Kafka topics for each Knative Channel
+targeting the Bus (named `<namespace>.<channel-name>`.
+Clients should avoid interacting with topics provisioned by the bus.
+
+The dispatcher
+- receives events via a Channel's Service from inside the cluster and
+writes them to the corresponding Kafka topic
+- creates a Kafka consumer for each `Subscription`, that reads events
+from the subscription's channel and forwards them over HTTP to the
+subscriber.
+
+To view logs:
+- for the dispatcher `kail -d kafka-bus -c dispatcher`
+- for the provisioner `kail -d kafka-bus-provisioner -c provisioner`

config/buses/kafka/kafka-broker.yaml

@@ -0,0 +1,89 @@
+########################################## KAFKA BROKER ######################################
+# The following does not need to live in the same namespace as the bus.
+---
+apiVersion: extensions/v1beta1
+kind: Deployment
+metadata:
+  name: kafka-broker
+spec:
+  replicas: 1
+  template:
+    metadata:
+      labels:
+        app: kafka-broker
+    spec:
+      containers:
+      - name: kafka-broker
+        image: wurstmeister/kafka:1.1.0
+        ports:
+        - containerPort: 9092
+        env:
+          - name: MY_POD_NAMESPACE
+            valueFrom:
+              fieldRef:
+                fieldPath: metadata.namespace
+          - name: KAFKA_BROKER_ID
+            value: "0"
+          - name: KAFKA_LISTENERS
+            value: "INTERNAL://:9093,EXTERNAL://:9092"
+          - name: KAFKA_ADVERTISED_LISTENERS
+            value: "INTERNAL://:9093,EXTERNAL://kafkabroker.$(MY_POD_NAMESPACE):9092"
+          - name: KAFKA_LISTENER_SECURITY_PROTOCOL_MAP
+            value: "INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT"
+          - name: KAFKA_INTER_BROKER_LISTENER_NAME
+            value: "INTERNAL"
+          - name: KAFKA_ZOOKEEPER_CONNECT
+            value: "zookeeper.$(MY_POD_NAMESPACE):2181"
+          - name: KAFKA_AUTO_CREATE_TOPICS_ENABLE
+            value: "false"
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: kafkabroker
+spec:
+  type: NodePort
+  selector:
+    app: kafka-broker
+  ports:
+  - port: 9092
+    name: kafka
+    protocol: TCP
+    nodePort: 31349
+
+---
+apiVersion: extensions/v1beta1
+kind: Deployment
+metadata:
+  name: zookeeper
+spec:
+  replicas: 1
+  template:
+    metadata:
+      labels:
+        app: zookeeper
+    spec:
+      containers:
+      - name: zookeeper
+        image: wurstmeister/zookeeper:3.4.6
+        ports:
+        - containerPort: 2181
+        env:
+        - name: ZOOKEEPER_ID
+          value: "1"
+        - name: ZOOKEEPER_SERVER_1
+          value: zookeeper
+
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: zookeeper
+spec:
+  selector:
+    app: zookeeper
+  ports:
+  - port: 2181
+    name: zookeeper
+    protocol: TCP
+

config/buses/kafka/kafka-bus.yaml

@@ -0,0 +1,30 @@
+apiVersion: channels.knative.dev/v1alpha1
+kind: Bus
+metadata:
+  name: kafka
+spec:
+  parameters:
+    subscription:
+    - name: "initialOffset"
+      description: "The initial offset to use when subscribing, either Oldest or Newest. Defaults to Newest."
+      default: "Newest"
+  provisioner:
+    name: provisioner
+    image: github.com/knative/eventing/pkg/buses/kafka/provisioner
+    args: [
+      "-logtostderr",
+      "-stderrthreshold", "INFO",
+    ]
+    env:
+    - name: KAFKA_BROKERS
+      value: "kafkabroker.kafka:9092"
+  dispatcher:
+    name: dispatcher
+    image: github.com/knative/eventing/pkg/buses/kafka/dispatcher
+    args: [
+      "-logtostderr",
+      "-stderrthreshold", "INFO",
+    ]
+    env:
+    - name: KAFKA_BROKERS
+      value: "kafkabroker.kafka:9092"