Add support for KIP-429 Incremental Rebalancing

Also fixes consumer.Close() that previously would not trigger the rebalance
callback leaving the application no way to do its revoke logic, if any.
The fix is to Unsubscribe() first, which triggers the rebalance, and then
Close().

Author: edenhill

CHANGELOG.md

@@ -2,8 +2,11 @@
 
 ## v1.6.0
 
-v1.6.0 is a feature release adding support for KIP-429 Incremental rebalancing.
+v1.6.0 is a feature release adding support for KIP-429 Incremental rebalancing
+and KIP-480 Sticky producer partitioner.
 
+ - See [cooperative_consumer_example.go](examples/cooperative_consumer_example/cooperative_consumer_example.go)
+   for an example how to use the new incremental rebalancing consumer.
  - Bundles librdkafka v1.6.0 - see release notes for all enhancements and fixes.
 
 

examples/cooperative_consumer_example/.gitignore

@@ -0,0 +1 @@
+cooperative_consumer_example

examples/cooperative_consumer_example/cooperative_consumer_example.go

@@ -0,0 +1,155 @@
+// Example high-level Apache Kafka consumer using the
+// cooperative incremental rebalancing protocol which allows
+// seamless reassignment of partitions to other group members.
+package main
+
+/**
+ * Copyright 2020 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.
+ */
+
+import (
+	"fmt"
+	"github.com/confluentinc/confluent-kafka-go/kafka"
+	"os"
+	"os/signal"
+	"syscall"
+)
+
+func main() {
+
+	if len(os.Args) < 4 {
+		fmt.Fprintf(os.Stderr, "Usage: %s <broker> <group> <topics..>\n",
+			os.Args[0])
+		os.Exit(1)
+	}
+
+	broker := os.Args[1]
+	group := os.Args[2]
+	topics := os.Args[3:]
+	sigchan := make(chan os.Signal, 1)
+	signal.Notify(sigchan, syscall.SIGINT, syscall.SIGTERM)
+
+	c, err := kafka.NewConsumer(&kafka.ConfigMap{
+		"bootstrap.servers": broker,
+		// Avoid connecting to IPv6 brokers:
+		// This is needed for the ErrAllBrokersDown show-case below
+		// when using localhost brokers on OSX, since the OSX resolver
+		// will return the IPv6 addresses first.
+		// You typically don't need to specify this configuration
+		// property.
+		"broker.address.family": "v4",
+		// Consumer group ID
+		"group.id": group,
+		// Use the cooperative incremental rebalance protocol.
+		"partition.assignment.strategy": "cooperative-sticky",
+		// Start reading from the first message of each assigned
+		// partition if there are no previously committed offsets
+		// for this group.
+		"auto.offset.reset": "earliest"})
+
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Failed to create consumer: %s\n", err)
+		os.Exit(1)
+	}
+
+	fmt.Printf("Created Consumer %v\n", c)
+
+	// Subscribe to topics, call the rebalanceCallback on assignment/revoke.
+	// The rebalanceCallback is triggered from c.Poll() and c.Close() below.
+	err = c.SubscribeTopics(topics, rebalanceCallback)
+
+	run := true
+
+	for run == true {
+		select {
+		case sig := <-sigchan:
+			fmt.Printf("Caught signal %v: terminating\n", sig)
+			run = false
+		default:
+			ev := c.Poll(100)
+			if ev == nil {
+				continue
+			}
+
+			switch e := ev.(type) {
+			case *kafka.Message:
+				fmt.Printf("%% Message on %s:\n%s\n",
+					e.TopicPartition, string(e.Value))
+				if e.Headers != nil {
+					fmt.Printf("%% Headers: %v\n", e.Headers)
+				}
+			case kafka.Error:
+				// Errors should generally be considered
+				// informational, the client will try to
+				// automatically recover.
+				// But in this example we choose to terminate
+				// the application if all brokers are down.
+				fmt.Fprintf(os.Stderr, "%% Error: %v: %v\n",
+					e.Code(), e)
+				if e.Code() == kafka.ErrAllBrokersDown {
+					run = false
+				}
+			default:
+				fmt.Printf("Ignored %v\n", e)
+			}
+		}
+	}
+
+	fmt.Printf("Closing consumer\n")
+	c.Close()
+}
+
+// rebalanceCallback is called on each group rebalance to assign additional
+// partitions, or remove existing partitions, from the consumer's current
+// assignment.
+//
+// The application may use this optional callback to inspect the assignment,
+// alter the initial start offset (the .Offset field of each assigned partition),
+// and read/write offsets to commit to an alternative store outside of Kafka.
+func rebalanceCallback(c *kafka.Consumer, event kafka.Event) error {
+
+	switch ev := event.(type) {
+	case kafka.AssignedPartitions:
+		fmt.Fprintf(os.Stderr,
+			"%% %s rebalance: %d new partition(s) assigned: %v\n",
+			c.GetRebalanceProtocol(), len(ev.Partitions),
+			ev.Partitions)
+
+		// The application may update the start .Offset of each
+		// assigned partition and then call IncrementalAssign().
+		// Even though this example does not alter the offsets we
+		// provide the call to IncrementalAssign() as an example.
+		err := c.IncrementalAssign(ev.Partitions)
+		if err != nil {
+			panic(err)
+		}
+
+	case kafka.RevokedPartitions:
+		fmt.Fprintf(os.Stderr,
+			"%% %s rebalance: %d partition(s) revoked: %v\n",
+			c.GetRebalanceProtocol(), len(ev.Partitions),
+			ev.Partitions)
+		if c.AssignmentLost() {
+			// Our consumer has been kicked out of the group and the
+			// entire assignment is thus lost.
+			fmt.Fprintf(os.Stderr, "%% Current assignment lost!\n")
+		}
+
+		// The client automatically calls IncrementalUnassign() unless
+		// the callback has already called that method.
+	}
+
+	return nil
+}