Expose the librdkafka stats as events in the go client (confluentinc#57)

Author: hqin

kafka/consumer.go

@@ -317,7 +317,7 @@ func NewConsumer(conf *ConfigMap) (*Consumer, error) {
 	cErrstr := (*C.char)(C.malloc(C.size_t(256)))
 	defer C.free(unsafe.Pointer(cErrstr))
 
-	C.rd_kafka_conf_set_events(cConf, C.RD_KAFKA_EVENT_REBALANCE|C.RD_KAFKA_EVENT_OFFSET_COMMIT)
+	C.rd_kafka_conf_set_events(cConf, C.RD_KAFKA_EVENT_REBALANCE|C.RD_KAFKA_EVENT_OFFSET_COMMIT|C.RD_KAFKA_EVENT_STATS)
 
 	c.handle.rk = C.rd_kafka_new(C.RD_KAFKA_CONSUMER, cConf, cErrstr, 256)
 	if c.handle.rk == nil {

kafka/event.go

@@ -55,6 +55,15 @@ type Event interface {
 
 // Specific event types
 
+// Stats statistics event
+type Stats struct {
+	statsJSON string
+}
+
+func (e Stats) String() string {
+	return e.statsJSON
+}
+
 // AssignedPartitions consumer group rebalance event: assigned partition set
 type AssignedPartitions struct {
 	Partitions []TopicPartition
@@ -190,6 +199,9 @@ out:
 				retval = newErrorFromCString(cErr, C.rd_kafka_event_error_string(rkev))
 			}
 
+		case C.RD_KAFKA_EVENT_STATS:
+			retval = &Stats{C.GoString(C.rd_kafka_event_stats(rkev))}
+
 		case C.RD_KAFKA_EVENT_DR:
 			// Producer Delivery Report event
 			// Each such event contains delivery reports for all

kafka/event_test.go

@@ -37,4 +37,7 @@ func TestEventAPIs(t *testing.T) {
 
 	committedOffsets := OffsetsCommitted{}
 	t.Logf("%s\n", committedOffsets.String())
+
+	stats := Stats{"{\"name\": \"Producer-1\"}"}
+	t.Logf("Stats: %s\n", stats.String())
 }

kafka/producer.go

@@ -275,7 +275,7 @@ func NewProducer(conf *ConfigMap) (*Producer, error) {
 	cErrstr := (*C.char)(C.malloc(C.size_t(256)))
 	defer C.free(unsafe.Pointer(cErrstr))
 
-	C.rd_kafka_conf_set_events(cConf, C.RD_KAFKA_EVENT_DR)
+	C.rd_kafka_conf_set_events(cConf, C.RD_KAFKA_EVENT_DR|C.RD_KAFKA_EVENT_STATS)
 
 	// Create librdkafka producer instance
 	p.handle.rk = C.rd_kafka_new(C.RD_KAFKA_PRODUCER, cConf, cErrstr, 256)

kafka/stats_event_test.go

@@ -0,0 +1,131 @@
+/**
+ * Copyright 2017 Confluent Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka
+
+import (
+	"encoding/json"
+	"testing"
+	"time"
+)
+
+// handleStatsEvent checks for stats event and signals on statsReceived
+func handleStatsEvent(t *testing.T, eventCh chan Event, statsReceived chan bool) {
+	for ev := range eventCh {
+		switch e := ev.(type) {
+		case *Stats:
+			t.Logf("Stats: %v", e)
+
+			// test if the stats string can be decoded into JSON
+			var raw map[string]interface{}
+			err := json.Unmarshal([]byte(e.String()), &raw) // convert string to json
+			if err != nil {
+				t.Fatalf("json unmarshall error: %s", err)
+			}
+			t.Logf("Stats['name']: %s", raw["name"])
+			close(statsReceived)
+			return
+		default:
+			t.Logf("Ignored event: %v", e)
+		}
+	}
+}
+
+// TestStatsEventProducerFunc dry-test stats event, no broker is needed.
+func TestStatsEventProducerFunc(t *testing.T) {
+	testProducerFunc(t, false)
+}
+
+func TestStatsEventProducerChannel(t *testing.T) {
+	testProducerFunc(t, true)
+}
+
+func testProducerFunc(t *testing.T, withProducerChannel bool) {
+
+	p, err := NewProducer(&ConfigMap{
+		"statistics.interval.ms": 50,
+		"socket.timeout.ms":      10,
+		"default.topic.config":   ConfigMap{"message.timeout.ms": 10}})
+	if err != nil {
+		t.Fatalf("%s", err)
+	}
+	defer p.Close()
+
+	t.Logf("Producer %s", p)
+
+	topic1 := "gotest"
+
+	// go routine to check for stats event
+	statsReceived := make(chan bool)
+	go handleStatsEvent(t, p.Events(), statsReceived)
+
+	if withProducerChannel {
+		err = p.Produce(&Message{TopicPartition: TopicPartition{Topic: &topic1, Partition: 0},
+			Value: []byte("Own drChan"), Key: []byte("This is my key")},
+			nil)
+		if err != nil {
+			t.Errorf("Produce failed: %s", err)
+		}
+	} else {
+		p.ProduceChannel() <- &Message{TopicPartition: TopicPartition{Topic: &topic1, Partition: 0},
+			Value: []byte("Own drChan"), Key: []byte("This is my key")}
+
+	}
+
+	select {
+	case <-statsReceived:
+		t.Logf("Stats recevied")
+	case <-time.After(time.Second * 3):
+		t.Fatalf("Excepted stats but got none")
+	}
+
+	return
+}
+
+// TestStatsEventConsumerChannel dry-tests stats event for consumer, no broker is needed.
+func TestStatsEventConsumerChannel(t *testing.T) {
+
+	c, err := NewConsumer(&ConfigMap{
+		"group.id":                 "gotest",
+		"statistics.interval.ms":   50,
+		"go.events.channel.enable": true,
+		"socket.timeout.ms":        10,
+		"session.timeout.ms":       10})
+	if err != nil {
+		t.Fatalf("%s", err)
+	}
+
+	defer c.Close()
+
+	t.Logf("Consumer %s", c)
+
+	// go routine to check for stats event
+	statsReceived := make(chan bool)
+	go handleStatsEvent(t, c.Events(), statsReceived)
+
+	err = c.Subscribe("gotest", nil)
+	if err != nil {
+		t.Errorf("Subscribe failed: %s", err)
+	}
+
+	select {
+	case <-statsReceived:
+		t.Logf("Stats recevied")
+	case <-time.After(time.Second * 3):
+		t.Fatalf("Excepted stats but got none")
+	}
+
+}