metricbeat: fix the `consumergroup` of kafka module incorrectly work #5880

wangdisdu · 2017-12-14T12:04:57Z

The consumergroup fetch offset info by sending OffsetFetch request to Kafka.
But the topic and partition parameters missed in the sent OffsetFetch request.
These parameters is required, and kafka will do not response anything if they are missed.

This is Kafka protocol guide: http://kafka.apache.org/protocol.html#The_Messages_OffsetFetch

elasticmachine · 2017-12-14T12:04:58Z

Can one of the admins verify this patch?

houndci-bot · 2017-12-14T12:05:05Z

metricbeat/module/kafka/consumergroup/query.go

+		_partitions := make(map[string][]int32)
+		for topic, partitions := range topics {
+			if topicsFilter == nil || topicsFilter(topic) {
+				for partitionID, _ := range partitions {


should omit 2nd value from range; this loop is equivalent to for partitionID := range ...

houndci-bot · 2017-12-14T12:05:05Z

metricbeat/module/kafka/broker.go

@@ -230,11 +230,16 @@ func (b *Broker) DescribeGroups(
 	return groups, nil
 }

-func (b *Broker) FetchGroupOffsets(group string) (*sarama.OffsetFetchResponse, error) {
+func (b *Broker) FetchGroupOffsets(group string, partitions map[string][]int32) (*sarama.OffsetFetchResponse, error) {


exported method Broker.FetchGroupOffsets should have comment or be unexported

houndci-bot · 2017-12-14T12:05:05Z

metricbeat/module/kafka/consumergroup/query.go

+		_partitions := make(map[string][]int32)
+		for topic, partitions := range topics {
+			if topicsFilter == nil || topicsFilter(topic) {
+				for partitionID, _ := range partitions {


should omit 2nd value from range; this loop is equivalent to for partitionID := range ...

houndci-bot · 2017-12-14T12:05:06Z

metricbeat/module/kafka/broker.go

@@ -230,11 +230,16 @@ func (b *Broker) DescribeGroups(
 	return groups, nil
 }

-func (b *Broker) FetchGroupOffsets(group string) (*sarama.OffsetFetchResponse, error) {
+func (b *Broker) FetchGroupOffsets(group string, partitions map[string][]int32) (*sarama.OffsetFetchResponse, error) {


exported method Broker.FetchGroupOffsets should have comment or be unexported

houndci-bot · 2017-12-15T02:30:11Z

metricbeat/module/kafka/consumergroup/query.go

-			return err
+	for waiting > 0 {
+		ret := <-results
+		waiting -= 1


should replace waiting -= 1 with waiting--

houndci-bot · 2017-12-15T02:30:11Z

metricbeat/module/kafka/consumergroup/query.go


 		wg.Add(1)
-		go func() {
+		waiting += 1


should replace waiting += 1 with waiting++

ruflin · 2017-12-29T01:57:06Z

@wangdisdu Just wanted to let you know that we saw the PR. As @urso was thinking already a bit about how to fix Kafka I'm hoping he can have a look at it.

urso

Thanks for contributing to the kafka module. It's a great way forword. Unfortunately I don't think it fixes all issues we have: basically some consumergroup state being held by another kafka broker.

urso · 2018-01-04T11:20:29Z

metricbeat/module/kafka/consumergroup/query.go

+
+	waiting := 0
+	for group, topics := range assignments {
+		_partitions := make(map[string][]int32)


no underscore please.

in beats codebase we prefer to use make/new only if there is not other syntax pattern. Use

partitions = map[string][]int32{}

How about (reduce key lookups and appends/alloc to slice):

queryTopics := map[string][]int32{} for topic, partitions := range topics { if topicFilter != nil && !topicFilter(topic) { continue } // copy partition ids L := len(partitions) if L == 0 { continue } ids, i := make([]int32, L), 0 for partition := range partitions { ids[i], i = partition, i+1 } queryTopics[topic] = ids } if len(queryTopics) == 0 { continue } ...

fixed: change underscore variable name

urso · 2018-01-04T11:26:56Z

metricbeat/module/kafka/consumergroup/query.go

+				for partitionID := range partitions {
+					if _, ok := _partitions[topic]; !ok {
+						_partitions[topic] = make([]int32, 0)
+					}


This if-clause is not required. _partitions[topic] returns nil, if topic is not known. append(([]int32)(nil), value) returns a new []int32{value}.

fixed: optimalized the code generate partition array

urso · 2018-01-04T11:39:15Z

metricbeat/module/kafka/consumergroup/query.go

-	results := make(chan result, len(groups))
-	for _, group := range groups {
-		group := group
+	results := make(chan result, len(assignments))


As the loop collecting data does not wait for wg.Wait anymore, but through all results, the wg is not really required anymore. We can even introduce an unbuffered channel make(chan result) now.

fixed: removed unecessary waitgroup

urso · 2018-01-04T11:41:45Z

metricbeat/module/kafka/consumergroup/query.go

+		ret := <-results
+		waiting--
+		if ret.err != nil {
+			continue


originally we did return an error -> we still want to return the first error we encounter

for waiting > 0 { ret := <- results waiting-- if ret.err != nil { if err == nil { err = ret.err } continue } ... } if err != nil { return err }

fixed: return the error if has

wangdisdu · 2018-01-05T03:54:10Z

Hi @ruflin @urso Thanks for your reviewed. I have adjusted this PR:

removed unecessary waitgroup
change underscore variable name
optimalized the code generate partition array
return the error if has

And For the issue: some consumergroup state being held by another kafka broker. I do not think it is an issue. Because the metricbeat will collect consumergroup info only from configed kafka host. If someone want collect all info, he need to config all kafka host in config file.

urso · 2018-01-05T10:44:02Z

metricbeat/module/kafka/consumergroup/query.go

 			}
-			return err
+			continue


we can stop processing data after having seen an error by doing this:

for waiting > 0 { ret := <-results waiting-- if ret.err != nil && err == nil { err = ret.err } if err != nil { continue } ... }

This will keep the waiting loop alive, assign the first ret.err only, but also stop processing any further results (results are discarded anyways).

@urso I fix it.

urso · 2018-01-05T10:50:40Z

And For the issue: some consumergroup state being held by another kafka broker. I do not think it is an issue. Because the metricbeat will collect consumergroup info only from configed kafka host. If someone want collect all info, he need to config all kafka host in config file.

This might still be a problem. The consumergroup needs to correlate data, which are potentially distributed between different kafka nodes. Problem is, the metricsets in metricbeat are fully isolated. That is, even if you configure all brokers, the metricsets are running in isolation per broker -> can not correlated meta-data distributed on different brokers.

See related ticket: #4285
The discuss forums discussion linked in the ticket contains quite some more details.

wangdisdu · 2018-01-11T12:25:51Z

@urso Hi, I found out that it is not a problem.

The offsets for a given consumer group are maintained by a specific broker called the group coordinator. Consumer needs to fetch offset from this specific broker. Link: kafka protocol guide

It means meta data of consumer group stored in one specific broker called coordinator.
We get the group list from a coordinator first, and can fetch all offsets of one group info from it.

And I test and verify it on kafka 0.10.2.
I got all offset info of group "wangdi-test-mb-group" from one broker "192.168.31.31:9092":

2018-01-11T20:02:36.540+0800	DEBUG	[publish]	pipeline/processor.go:275	Publish event: {
  "@timestamp": "2018-01-11T12:02:36.531Z",
  "@metadata": {
    "beat": "metricbeat",
    "type": "doc",
    "version": "7.0.0-alpha1"
  },
  "metricset": {
    "name": "consumergroup",
    "module": "kafka",
    "host": "192.168.31.31:9092",
    "rtt": 8546
  },
  "kafka": {
    "consumergroup": {
      "offset": 16692,
      "meta": "",
      "broker": {
        "id": 2,
        "address": "192.168.31.31:9092"
      },
      "topic": "wangdi-test-mb-topic",
      "partition": 0,
      "client": {
        "id": "consumer-1",
        "host": "172.16.128.23",
        "member_id": "consumer-1-ed7ae7c7-0534-49b2-84c9-26b8c64063df"
      },
      "id": "wangdi-test-mb-group"
    }
  },
  "beat": {
    "version": "7.0.0-alpha1",
    "name": "localhost",
    "hostname": "localhost"
  }
}
2018-01-11T20:02:36.540+0800	DEBUG	[publish]	pipeline/processor.go:275	Publish event: {
  "@timestamp": "2018-01-11T12:02:36.531Z",
  "@metadata": {
    "beat": "metricbeat",
    "type": "doc",
    "version": "7.0.0-alpha1"
  },
  "kafka": {
    "consumergroup": {
      "id": "wangdi-test-mb-group",
      "offset": 16429,
      "topic": "wangdi-test-mb-topic",
      "partition": 1,
      "meta": "",
      "broker": {
        "id": 2,
        "address": "192.168.31.31:9092"
      },
      "client": {
        "id": "consumer-1",
        "host": "172.16.128.23",
        "member_id": "consumer-1-ed7ae7c7-0534-49b2-84c9-26b8c64063df"
      }
    }
  },
  "metricset": {
    "name": "consumergroup",
    "module": "kafka",
    "host": "192.168.31.31:9092",
    "rtt": 8566
  },
  "beat": {
    "name": "localhost",
    "hostname": "localhost",
    "version": "7.0.0-alpha1"
  }
}
2018-01-11T20:02:36.540+0800	DEBUG	[publish]	pipeline/processor.go:275	Publish event: {
  "@timestamp": "2018-01-11T12:02:36.531Z",
  "@metadata": {
    "beat": "metricbeat",
    "type": "doc",
    "version": "7.0.0-alpha1"
  },
  "metricset": {
    "name": "consumergroup",
    "module": "kafka",
    "host": "192.168.31.31:9092",
    "rtt": 8574
  },
  "kafka": {
    "consumergroup": {
      "partition": 2,
      "offset": 16499,
      "broker": {
        "id": 2,
        "address": "192.168.31.31:9092"
      },
      "id": "wangdi-test-mb-group",
      "meta": "",
      "client": {
        "member_id": "consumer-1-ed7ae7c7-0534-49b2-84c9-26b8c64063df",
        "id": "consumer-1",
        "host": "172.16.128.23"
      },
      "topic": "wangdi-test-mb-topic"
    }
  },
  "beat": {
    "name": "localhost",
    "hostname": "localhost",
    "version": "7.0.0-alpha1"
  }
}

urso · 2018-01-11T13:26:48Z

@wangdisdu Cool. how many kafka brokers have you used? Did you use one central metricbeat, or one per host? Normally we recommend installing metricbeat on each host and configure metricbeat kafka module to connect to localhost broker only (metricbeat matches the meta-data with the hostname to figure out the actual brokers name from the metadata response). With having one metricbeat instance per host, we can also collect CPU/memory/disk usage on this host as well. Even though state is distributed between kafka nodes, each metricbeat will collect a subset of the overall kafka cluster state.

PR look good to me. Can you add a changelog about fixing the OffsetFetch request?

wangdisdu · 2018-01-12T01:34:11Z

@urso For test kafka metricset, I build a kafka cluster with 3 brokers.
For produce test data, I use standard java kafka client to produce and consume message.
I test metricbeat on per broker by configing local host ip (same as localhost).

And I added changelog.

…ot fetch anything The `consumergroup` fetch offset info by sending `OffsetFetch` request to Kafka. But the `topic` and `partition` parameters missed in the sent `OffsetFetch` request. These parameters is required, and kafka will do not response anything if they are missed.

wangdisdu · 2018-01-17T08:45:52Z

@ruflin Will this PR be merged? So I can do more thing base on it, like batching requests together for efficiency.

urso · 2018-01-17T20:52:54Z

@wangdisdu how many partitions and consumer groups did you use. The ticket I mentioned uses 10 partitions and 3 consumer groups, as sometimes for less consumer groups we could see all data. Only adding more groups/partitions to the picture to show us some missing consumer groups data.

urso · 2018-01-17T20:54:04Z

Merged the PR. Thank you for taking the time to work on this long outstanding issue. Your help is highly appreciated ;)

wangdisdu · 2018-01-18T03:12:08Z

@urso It working well even if add more groups. I test and verify it:

I create topic wd-test-mb-cg with config 10 partitions and 2 peplication factors
I use 5 consumer groups to fetch message from topic wd-test-mb-cg
I checked the data is not missing, the count of partition of every group is always 10
I increase count of consumer groups to 50 to fetch message from topic wd-test-mb-cg
I checked the data is not missing, the count of partition of every group is always 10

ruflin · 2018-01-18T22:47:18Z

@wangdisdu I can only double down on what @urso said. We really appreciate your work here and being patient with us to get the changes in. ❤️

npredey · 2018-06-20T19:11:51Z

Which version of metricbeat is this available on and will it work with Elasticsearch 5.3? Thanks.

ruflin · 2018-06-21T10:59:41Z

This is part of the 6.3 release.

houndci-bot reviewed Dec 14, 2017

View reviewed changes

wangdisdu changed the title ~~metricbeat: fix the consumergroup of kafka module do not fetch anything~~ metricbeat: fix the consumergroup of kafka module incorrectly work Dec 14, 2017

wangdisdu force-pushed the metricbeat-kafka-consumergroup branch 2 times, most recently from 8a04335 to 03d7f64 Compare December 15, 2017 02:30

houndci-bot reviewed Dec 15, 2017

View reviewed changes

wangdisdu force-pushed the metricbeat-kafka-consumergroup branch from 03d7f64 to e4d1877 Compare December 15, 2017 02:44

ruflin added Metricbeat Metricbeat review labels Dec 17, 2017

ruflin requested a review from urso December 17, 2017 10:36

ruflin added the module label Dec 29, 2017

urso suggested changes Jan 4, 2018

View reviewed changes

wangdisdu force-pushed the metricbeat-kafka-consumergroup branch from e4d1877 to 664fb09 Compare January 5, 2018 03:45

urso reviewed Jan 5, 2018

View reviewed changes

urso approved these changes Jan 5, 2018

View reviewed changes

wangdisdu force-pushed the metricbeat-kafka-consumergroup branch from 664fb09 to f0c00d1 Compare January 5, 2018 12:27

wangdisdu force-pushed the metricbeat-kafka-consumergroup branch from f0c00d1 to f07636a Compare January 12, 2018 01:51

urso merged commit da70593 into elastic:master Jan 17, 2018

urso added the needs_backport PR is waiting to be backported to other branches. label Feb 2, 2018

urso removed the needs_backport PR is waiting to be backported to other branches. label Jun 18, 2018

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

metricbeat: fix the `consumergroup` of kafka module incorrectly work #5880

metricbeat: fix the `consumergroup` of kafka module incorrectly work #5880

wangdisdu commented Dec 14, 2017

elasticmachine commented Dec 14, 2017

houndci-bot Dec 14, 2017

houndci-bot Dec 14, 2017

houndci-bot Dec 14, 2017

houndci-bot Dec 14, 2017

houndci-bot Dec 15, 2017

houndci-bot Dec 15, 2017

ruflin commented Dec 29, 2017

urso left a comment

urso Jan 4, 2018 •

edited

Loading

wangdisdu Jan 5, 2018 •

edited

Loading

urso Jan 4, 2018

wangdisdu Jan 5, 2018

urso Jan 4, 2018

wangdisdu Jan 5, 2018 •

edited

Loading

urso Jan 4, 2018

wangdisdu Jan 5, 2018

wangdisdu commented Jan 5, 2018 •

edited

Loading

urso Jan 5, 2018 •

edited

Loading

wangdisdu Jan 5, 2018

urso commented Jan 5, 2018

wangdisdu commented Jan 11, 2018 •

edited

Loading

urso commented Jan 11, 2018

wangdisdu commented Jan 12, 2018 •

edited

Loading

wangdisdu commented Jan 17, 2018

urso commented Jan 17, 2018

urso commented Jan 17, 2018

wangdisdu commented Jan 18, 2018 •

edited

Loading

ruflin commented Jan 18, 2018

npredey commented Jun 20, 2018

ruflin commented Jun 21, 2018

metricbeat: fix the consumergroup of kafka module incorrectly work #5880

metricbeat: fix the consumergroup of kafka module incorrectly work #5880

Conversation

wangdisdu commented Dec 14, 2017

elasticmachine commented Dec 14, 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

ruflin commented Dec 29, 2017

urso left a comment

Choose a reason for hiding this comment

urso Jan 4, 2018 • edited Loading

Choose a reason for hiding this comment

wangdisdu Jan 5, 2018 • edited Loading

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

wangdisdu Jan 5, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wangdisdu commented Jan 5, 2018 • edited Loading

urso Jan 5, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

urso commented Jan 5, 2018

wangdisdu commented Jan 11, 2018 • edited Loading

urso commented Jan 11, 2018

wangdisdu commented Jan 12, 2018 • edited Loading

wangdisdu commented Jan 17, 2018

urso commented Jan 17, 2018

urso commented Jan 17, 2018

wangdisdu commented Jan 18, 2018 • edited Loading

ruflin commented Jan 18, 2018

npredey commented Jun 20, 2018

ruflin commented Jun 21, 2018

metricbeat: fix the `consumergroup` of kafka module incorrectly work #5880

metricbeat: fix the `consumergroup` of kafka module incorrectly work #5880

urso Jan 4, 2018 •

edited

Loading

wangdisdu Jan 5, 2018 •

edited

Loading

wangdisdu Jan 5, 2018 •

edited

Loading

wangdisdu commented Jan 5, 2018 •

edited

Loading

urso Jan 5, 2018 •

edited

Loading

wangdisdu commented Jan 11, 2018 •

edited

Loading

wangdisdu commented Jan 12, 2018 •

edited

Loading

wangdisdu commented Jan 18, 2018 •

edited

Loading