Expose the librdkafka stats as events in the go client #57

hqin · 2017-04-15T00:15:34Z

@edenhill Please review the changes along with the librdkafka PR confluentinc/librdkafka#1171 ("added support for stats as events")

edenhill · 2017-04-19T18:21:11Z

kafka/integration_test.go

+		if mt.msgcnt == int64(cap(mt.msgs)) {
+			mt.t.Logf("***** Buffer full. not saving message: %v\n", e)
+		} else {
+			// mt.t.Logf("Saving message to mt.msgs[%d]: %v\n", mt.msgcnt, e)


remove comment

edenhill · 2017-04-19T18:21:54Z

kafka/integration_test.go

+		} else {
+			// mt.t.Logf("Saving message to mt.msgs[%d]: %v\n", mt.msgcnt, e)
+			mt.msgs[mt.msgcnt] = e
+			mt.msgcnt++


is msgcnt needed, isn't len(mt.msgs) sufficient?

edenhill · 2017-04-19T18:22:45Z

kafka/integration_test.go

 		}
 	case PartitionEOF:
 		break // silence
+	case *Stats:
+		mt.t.Logf("Stats: %v", e)


The JSON should be verified to be syntactically correct, if there is an easy way to do that with Go without requiring full-blow corresponding go types

edenhill · 2017-04-19T18:22:54Z

kafka/integration_test.go

 		mt.t.Fatalf("Consumer error: %v", e)
+		//mt.t.Logf("Ignored event: %v\n", e)


remove comment

edenhill · 2017-04-19T18:25:27Z

kafka/integration_test.go

 	default:
+		// there are other unhandled events such as OffsetsCommitted


is this meant as a warning or fixme, ..?
Maybe we should handle it instead.

edenhill · 2017-04-19T18:28:07Z

kafka/integration_test.go

@@ -252,7 +285,9 @@ func producerTest(t *testing.T, testname string, testmsgs []*testmsgType, pc pro
 	}

 	mt := msgtrackerStart(t, len(testmsgs))
-
+	if pc.withStats {


can't this be set along with statistics.interval.ms at line ~276?

edenhill · 2017-04-19T18:28:52Z

kafka/integration_test.go

+
+	// verify that stats events are received
+	if pc.withStats {
+		if mt.statscnt == 0 {


combine into one if statement

edenhill · 2017-04-19T18:29:15Z

kafka/integration_test.go

@@ -351,7 +403,9 @@ func consumerTest(t *testing.T, testname string, msgcnt int, cc consumerCtrl, co

 	expCnt := msgcnt
 	mt := msgtrackerStart(t, expCnt)
-
+	if cc.withStats {
+		mt.withStats = true


edenhill · 2017-04-19T18:50:15Z

kafka/integration_test.go

+	if cc.withStats {
+		//wait for a while for stats to show up
+		for i := 0; i < 3; i++ {
+			if mt.statscnt != 0 {


Is this thread-safe? mt.statscnt is updated through another go-routine, right?
I suggest changing this to a channel instead to signal completion.
E.g., let the stats receiving go-routine count the number of stats and send a "done" message on a channel when it is done, which is waited on here (with a reasonable timeout).

edenhill · 2017-04-19T18:50:59Z

kafka/integration_test.go

@@ -552,6 +622,18 @@ func TestProducerFuncDR(t *testing.T) {
 		})
 }

+// test producer function-based API with stats and delivery report


As with librdkafka I'd like to see a unittest for the stats that doesnt rely on broker connectivity.

edenhill · 2017-04-19T18:51:32Z

This looks great! Just a couple of things needs sorting out

edenhill · 2017-05-02T17:55:00Z

@hqin Can you rebase and squash/fixup your commits to a single one?
Since the follow-up commit is quite large it makes it easier to review the full thing atomically.

Something like:

 $ git checkout master
 $ git pull origin master
 $ git checkout <yourbranch>
 $ git rebase -i master
 # in the editor, mark your second commit with "fixup" (or "f") and move it just below your first commit (if not already there) to silently squash it into your previous commit
 $ git log   # verify that there is only one commit
 # Force push
 $ git push --dry-run --force origin <yourbranch>   # not master!
 # when you think that looks good, remove --dry-run

edenhill · 2017-05-08T20:45:46Z

kafka/stats_event_test.go

+
+			// test if the stats string can be decoded into JSON
+			var raw map[string]interface{}
+			json.Unmarshal([]byte(e.String()), &raw) // convert string to json


Should check for err return here

edenhill · 2017-05-08T20:48:06Z

kafka/stats_event_test.go

+			Value: []byte("Own drChan"), Key: []byte("This is my key")}
+
+	}
+	p.Flush(2000)


Since no brokers are configured and the default message.timeout.ms is 5 minutes this flush(2s) won't do anything.

I suggest removing it and just relying on the timeout below.

removed p.Flush()

edenhill · 2017-05-08T20:48:22Z

kafka/stats_event_test.go

+func testProducerFunc(t *testing.T, withProducerChannel bool) {
+
+	p, err := NewProducer(&ConfigMap{
+		"statistics.interval.ms": 500,


any reason this is so "high"? 50 ms should suffice

Changed to 50ms

edenhill · 2017-05-11T08:29:19Z

LGTM.

freyayunfu · 2021-07-24T20:38:11Z

kafka/stats_event_test.go

+				t.Fatalf("json unmarshall error: %s", err)
+			}
+			t.Logf("Stats['name']: %s", raw["name"])
+			close(statsReceived)


@edenhill If the eventCh has more than one Stats messages, then statsReceived would be closed twice, which should panic the program. I wonder how this actually can work ? In my understanding, golang channel cannot be closed twice.

hqin requested a review from edenhill April 15, 2017 00:15

edenhill suggested changes Apr 19, 2017

View reviewed changes

hqin force-pushed the issue_32_stats_event branch from 887db23 to 0e63fb1 Compare May 3, 2017 16:59

edenhill suggested changes May 8, 2017

View reviewed changes

Expose the librdkafka stats as events in the go client

844faf0

hqin force-pushed the issue_32_stats_event branch from 0e63fb1 to 844faf0 Compare May 10, 2017 17:38

edenhill approved these changes May 11, 2017

View reviewed changes

edenhill merged commit 18b5a55 into confluentinc:master May 11, 2017

yxu-valleytider mentioned this pull request Apr 19, 2018

Expose stats through callback function #174

Closed

4 tasks

freyayunfu reviewed Jul 24, 2021

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Expose the librdkafka stats as events in the go client #57

Expose the librdkafka stats as events in the go client #57

hqin commented Apr 15, 2017

edenhill Apr 19, 2017

edenhill Apr 19, 2017

edenhill Apr 19, 2017

edenhill Apr 19, 2017

edenhill Apr 19, 2017

edenhill Apr 19, 2017

edenhill Apr 19, 2017

edenhill Apr 19, 2017

edenhill Apr 19, 2017

edenhill Apr 19, 2017

edenhill commented Apr 19, 2017

edenhill commented May 2, 2017

edenhill May 8, 2017

hqin May 10, 2017

edenhill May 8, 2017

hqin May 10, 2017

edenhill May 8, 2017

hqin May 10, 2017

edenhill commented May 11, 2017

freyayunfu Jul 24, 2021

		mt.t.Fatalf("Consumer error: %v", e)
		//mt.t.Logf("Ignored event: %v\n", e)

		default:
		// there are other unhandled events such as OffsetsCommitted

Expose the librdkafka stats as events in the go client #57

Expose the librdkafka stats as events in the go client #57

Conversation

hqin commented Apr 15, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

edenhill commented Apr 19, 2017

edenhill commented May 2, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

edenhill commented May 11, 2017

Choose a reason for hiding this comment