Merge pull request #28 from blendle/kb/librdkafka

Use the librdkafka implementation by confluent.

Author: jurre

Dockerfile

@@ -1,5 +1,17 @@
+FROM golang:alpine as builder
+
+WORKDIR /go/src/github.com/blendle/pg2kafka
+ADD . ./
+
+RUN apk --update --no-cache add git alpine-sdk bash
+RUN wget -qO- https://github.com/edenhill/librdkafka/archive/v0.11.4-RC1.tar.gz | tar xz
+RUN cd librdkafka-* && ./configure && make && make install
+RUN go get github.com/golang/dep/cmd/dep && dep ensure -vendor-only
+RUN go build -ldflags "-X main.version=$(git rev-parse --short @) -s -extldflags -static" -a -installsuffix cgo .
+
 FROM scratch
-MAINTAINER Jurre Stender <[email protected]>
+LABEL maintainer="Jurre Stender <[email protected]>"
 COPY sql ./sql
-COPY pg2kafka /
+COPY --from=builder /go/src/github.com/blendle/pg2kafka/pg2kafka /
+COPY --from=builder /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/ca-certificates.crt
 ENTRYPOINT ["/pg2kafka"]

Gopkg.lock

@@ -1,7 +1,3 @@
-[[constraint]]
-  branch = "master"
-  name = "github.com/blendle/go-streamprocessor"
-
 [[constraint]]
   branch = "master"
   name = "github.com/buger/jsonparser"

Gopkg.toml

@@ -157,12 +157,12 @@ kafka-topics \
 Then export the Kafka host as an URL so pg2kafka can use it:
 
 ```bash
-$ export KAFKA_PRODUCER_URL="kafka://localhost:9092"
+$ export KAFKA_BROKER="localhost:9092"
 ```
 
 ### Running the service locally
 
-Make sure you export the `DATABASE_URL` and `KAFKA_PRODUCER_URL`, and also
+Make sure you export the `DATABASE_URL` and `KAFKA_BROKER`, and also
 `export PERFORM_MIGRATIONS=true`.
 
 ```bash

README.md

@@ -10,12 +10,10 @@ import (
 	"time"
 
 	logger "github.com/blendle/go-logger"
-	"github.com/blendle/go-streamprocessor/stream"
-	"github.com/blendle/go-streamprocessor/streamclient"
-	"github.com/blendle/go-streamprocessor/streamclient/kafka"
-	"github.com/blendle/go-streamprocessor/streamclient/standardstream"
 	"github.com/blendle/pg2kafka/eventqueue"
+	"github.com/confluentinc/confluent-kafka-go/kafka"
 	"github.com/lib/pq"
+	"github.com/pkg/errors"
 	"go.uber.org/zap"
 )
 
@@ -24,6 +22,15 @@ var (
 	version        string
 )
 
+// Producer is the minimal required interface pg2kafka requires to produce
+// events to a kafka topic.
+type Producer interface {
+	Close()
+	Flush(int) int
+
+	Produce(*kafka.Message, chan kafka.Event) error
+}
+
 func main() {
 	conf := &logger.Config{
 		App:         "pg2kafka",
@@ -57,12 +64,8 @@ func main() {
 	}
 
 	producer := setupProducer()
-
-	defer func() {
-		if cerr := producer.Close(); cerr != nil {
-			logger.L.Fatal("Error closing producer", zap.Error(err))
-		}
-	}()
+	defer producer.Close()
+	defer producer.Flush(1000)
 
 	reportProblem := func(ev pq.ListenerEventType, err error) {
 		if err != nil {
@@ -91,7 +94,7 @@ func main() {
 
 // ProcessEvents queries the database for unprocessed events and produces them
 // to kafka.
-func ProcessEvents(p stream.Producer, eq *eventqueue.Queue) {
+func ProcessEvents(p Producer, eq *eventqueue.Queue) {
 	events, err := eq.FetchUnprocessedRecords()
 	if err != nil {
 		logger.L.Error("Error listening to pg", zap.Error(err))
@@ -100,7 +103,7 @@ func ProcessEvents(p stream.Producer, eq *eventqueue.Queue) {
 	produceMessages(p, events, eq)
 }
 
-func processQueue(p stream.Producer, eq *eventqueue.Queue) {
+func processQueue(p Producer, eq *eventqueue.Queue) {
 	pageCount, err := eq.UnprocessedEventPagesCount()
 	if err != nil {
 		logger.L.Fatal("Error selecting count", zap.Error(err))
@@ -113,7 +116,7 @@ func processQueue(p stream.Producer, eq *eventqueue.Queue) {
 
 func waitForNotification(
 	l *pq.Listener,
-	p stream.Producer,
+	p Producer,
 	eq *eventqueue.Queue,
 	signals chan os.Signal,
 ) {
@@ -134,39 +137,67 @@ func waitForNotification(
 	}
 }
 
-func produceMessages(p stream.Producer, events []*eventqueue.Event, eq *eventqueue.Queue) {
+func produceMessages(p Producer, events []*eventqueue.Event, eq *eventqueue.Queue) {
+	deliveryChan := make(chan kafka.Event)
 	for _, event := range events {
 		msg, err := json.Marshal(event)
 		if err != nil {
 			logger.L.Fatal("Error parsing event", zap.Error(err))
 		}
 
-		p.Messages() <- &stream.Message{
+		topic := topicName(event.TableName)
+		message := &kafka.Message{
+			TopicPartition: kafka.TopicPartition{
+				Topic:     &topic,
+				Partition: kafka.PartitionAny, // nolint: gotype
+			},
 			Value:     msg,
-			Topic:     topicName(event.TableName),
 			Key:       event.ExternalID,
 			Timestamp: event.CreatedAt,
 		}
-
+		if os.Getenv("DRY_RUN") != "" {
+			logger.L.Info("Would produce message", zap.Any("message", message))
+		} else {
+			err = p.Produce(message, deliveryChan)
+			if err != nil {
+				logger.L.Fatal("Failed to produce", zap.Error(err))
+			}
+			e := <-deliveryChan
+
+			result := e.(*kafka.Message)
+			if result.TopicPartition.Error != nil {
+				logger.L.Fatal("Delivery failed", zap.Error(result.TopicPartition.Error))
+			}
+		}
 		err = eq.MarkEventAsProcessed(event.ID)
 		if err != nil {
 			logger.L.Fatal("Error marking record as processed", zap.Error(err))
 		}
 	}
 }
 
-func setupProducer() stream.Producer {
-	options := func(sc *standardstream.Client, kc *kafka.Client) {
-		sc.Logger = logger.L
-		kc.Logger = logger.L
+func setupProducer() Producer {
+	broker := os.Getenv("KAFKA_BROKER")
+	if broker == "" {
+		panic("missing KAFKA_BROKER environment")
+	}
+
+	hostname, err := os.Hostname()
+	if err != nil {
+		hostname = os.Getenv("HOSTNAME")
 	}
 
-	producer, err := streamclient.NewProducer(options)
+	p, err := kafka.NewProducer(&kafka.ConfigMap{
+		"client.id":         hostname,
+		"bootstrap.servers": broker,
+		"partitioner":       "murmur2",
+		"compression.codec": "snappy",
+	})
 	if err != nil {
-		logger.L.Fatal("Unable to initialize producer", zap.Error(err))
+		panic(errors.Wrap(err, "failed to setup producer"))
 	}
 
-	return producer
+	return p
 }
 
 func topicName(tableName string) string {

main.go

@@ -5,8 +5,8 @@ import (
 	"os"
 	"testing"
 
-	"github.com/blendle/go-streamprocessor/streamclient/inmem"
 	"github.com/buger/jsonparser"
+	"github.com/confluentinc/confluent-kafka-go/kafka"
 
 	"github.com/blendle/pg2kafka/eventqueue"
 	_ "github.com/lib/pq"
@@ -54,23 +54,19 @@ func TestFetchUnprocessedRecords(t *testing.T) {
 		t.Fatalf("Error inserting events: %v", err)
 	}
 
-	opts := func(c *inmem.Client) {
-		c.ProducerTopic = "users"
+	p := &mockProducer{
+		messages: make([]*kafka.Message, 0),
 	}
-	s := inmem.NewStore()
-	pt := s.NewTopic("users")
-	c := inmem.NewClientWithStore(s, opts)
-	p := c.NewProducer()
 
 	ProcessEvents(p, eq)
 
 	expected := 4
-	actual := len(pt.Messages())
+	actual := len(p.messages)
 	if actual != expected {
 		t.Fatalf("Unexpected number of messages produced. Expected %d, got %d", expected, actual)
 	}
 
-	msg := pt.Messages()[0]
+	msg := p.messages[0]
 	email, err := jsonparser.GetString(msg.Value, "data", "email")
 	if err != nil {
 		t.Fatal(err)
@@ -89,7 +85,7 @@ func TestFetchUnprocessedRecords(t *testing.T) {
 		t.Errorf("Expected %v, got %v", "fefc72b4-d8df-4039-9fb9-bfcb18066a2b", externalID)
 	}
 
-	msg = pt.Messages()[3]
+	msg = p.messages[3]
 	email, err = jsonparser.GetString(msg.Value, "data", "email")
 	if err != nil {
 		t.Fatal(err)
@@ -99,8 +95,8 @@ func TestFetchUnprocessedRecords(t *testing.T) {
 		t.Errorf("Data did not match. Expected %v, got %v", "[email protected]", email)
 	}
 
-	if msg.Key != nil {
-		t.Errorf("Expected key to be nil, got %v", msg.Key)
+	if len(msg.Key) != 0 {
+		t.Errorf("Expected empty key, got %v", msg.Key)
 	}
 }
 
@@ -178,3 +174,20 @@ func TestParseTopicNamespace(t *testing.T) {
 		})
 	}
 }
+
+type mockProducer struct {
+	messages []*kafka.Message
+}
+
+func (p *mockProducer) Close() {
+}
+func (p *mockProducer) Flush(timeout int) int {
+	return 0
+}
+func (p *mockProducer) Produce(msg *kafka.Message, deliveryChan chan kafka.Event) error {
+	p.messages = append(p.messages, msg)
+	go func() {
+		deliveryChan <- msg
+	}()
+	return nil
+}

main_test.go

@@ -2,6 +2,4 @@
 set -eu
 
 GIT_COMMIT=$(git rev-parse --short @)
-export CGO_ENABLED=0
-
-go build -ldflags "-X main.version=$GIT_COMMIT -s -extldflags -static" -a -installsuffix cgo .
+docker build -t "eu.gcr.io/bnl-blendle/pg2kafka:$GIT_COMMIT" .