Confluent-kafka-go: Consumer.Close() never returns with multiple consumers in the same process

[dtheodor]

I'm running 2 consumers on the same topic and the same consumer group, using Poll() to consume messages. After having consumed messages with both of them, calling Close() on any of the consumers blocks indefinitely, reproducible 100% of the time. Debug logs show things being stuck at "waiting for rebalance_cb"

This is the gist with the go code to reproduce and all relevant information: https://gist.github.com/dtheodor/26821f951502aab2e325f860b459cbfc

This could be the same issue as https://github.com/confluentinc/confluent-kafka-go/issues/65 but you are talking about using the Events channel there, which I am not.

Calling Unassign() right before calling Close() allows Close() to return successfully, but having to call Unassign looks bogus since I never have to use Assign or Unassign in any other place, this is managed by the lib for me.

[edenhill]

Your code looks fine so I will try to reproduce and get back to you,

[edenhill]

What go client and librdkafka versions are you on?

[dtheodor]

go version go1.10.1 linux/amd64
librdkafka: 722175 0.11.4
confluent-kafka-go: v0.11.4

[dtheodor]

I can also reproduce the hang with just a single consumer if I add delay between the unassign and the close by calling (code is abbreviated)

// "enable.auto.commit":      true
// "auto.commit.interval.ms": 30 * 1000
c.Subscribe()
c.Poll()
c.Unassign()
time.Sleep(10 * time.Second)
c.Close() // this will never return

I need to do a c.Unsubscribe() before the Unassign for this to work

[edenhill]

You're only supposed to call Assign/Unassign from OnPartitionsAssigned/OnPartitionsRevoked callbacks/events.

[dtheodor]

So my workaround of calling Unassign before Close doesn't seem reliable

[dtheodor]

Reproducing with Unsubscribe, not using Unassign or time delays at all

// "enable.auto.commit":      true
// "auto.commit.interval.ms": 50 * 1000
c.Subscribe()
c.Poll()
c.Unsubscribe()
c.Close() // this will never return

logs keep outputing "waiting for rebalance_cb"

[dtheodor]

The issue is not reproducible with librdkafka 721407 0.11.1

[edenhill]

I can't reproduce this on master, nor v0.11.4 with the following program:

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,
        "group.id":                group,
        "enable.auto.commit":      true,
        "auto.commit.interval.ms": 50 * 1000,
    })

    if err != nil {
        fmt.Fprintf(os.Stderr, "Failed to create consumer: %s\n", err)
        os.Exit(1)
    }

    fmt.Printf("Created Consumer %v\n", c)

    err = c.SubscribeTopics(topics, nil)
    c.Poll(0)
    c.Unsubscribe()
    fmt.Printf("Closing\n")
    c.Close()
    fmt.Printf("Closed\n")
    os.Exit(0)
}

It terminates within half a second

[edenhill]

Trying your program now

[edenhill]

This issue is fixed in librdkafka adminapi branch which will be merged to master during this week and released in june.

The workaround for now is to drain the consumer queue by calling Poll() until nil is returned, which means all outstanding callbacks (and messages) have been processed.

[edenhill]

https://github.com/edenhill/librdkafka/commit/6c87c1305b7b5695ca506cf18164f69b063757d8

[agis]

The workaround for now is to drain the consumer queue by calling Poll() until nil is returned, which means all outstanding callbacks (and messages) have been processed.

@edenhill Maybe you mean Unsubscribe() Unassign() and then Poll() (since without Unassign(), Poll() might never return nil)?

[MaximGurschi]

@edenhill Is the proposed workaround in fact something one would want to do anyway?

[dtheodor]

FYI this is the full workaround implementation that works https://github.com/skroutz/rafka/blob/7e9dfb249ac8ff47594474b468aa57a2f21e4788/consumer.go#L67-L93

[MaximGurschi]

Thanks @dtheodor
However for the C interface i have to drain both queues?
rd_kafka_consumer_poll and rd_kafka_poll?

[MaximGurschi]

My current solution is to use ver 0.11.3.

[MaximGurschi]

FYI: The proposed workaround (unsubscribe plus drain) is unstable still.

[dtheodor]

It might be, were you able to deterministically reproduce the hang with it?

[MaximGurschi]

No, it was one of many attempts :/

[MaximGurschi]

When you call rd_kafka_consumer_close, the documentation specifies that there is no need to unsubscribe:
https://github.com/edenhill/librdkafka/wiki/Proper-termination-sequence

However, is there still a chance some messages were left in the internal queue and not processed?
Please correct me if i'm wrong, but if i want to process all messages that i can, before exiting, is it recommended to actually:

1. Unsubscribe
2. Drain consumer queue

, all before calling rd_kafka_consumer_close?

[edenhill]

Yes, that seems like a reasonable workaround for now.
We're looking into deprecating (and eventually removing) the channel-based consumer and producers due to these double-queue state issues, so the recommendation is to move to the functional API.

[MaximGurschi]

Hi Magnus, sorry not sure what post the reply is aimed at.

My question is not about the workaround but in general.

Is it correct that given an issue free version of librdkafka, if i simply call rd_kafka_consumer_close, that some messages might be left in the internal consumer queue?

And if that is the case, and i want to consume all messages that have made it from the broker to the consumer process that i should:

1. Unsubscribe
2. Drain consumer queue
3. rd_kafka_consumer_close

[edenhill]

Yes, messages will most likely be purged from the internal pre-fetch queue, but this is not a problem, these messages have not and are not committed, only the final message that your application has actually seen will be committed.

When you unsubscribe the pre-fetch queues are emptied, there is no way to finish processing queued messages without also fetching new messages, and there is typically little point to do so in a streaming environment.

[MaximGurschi]

So are you saying that if i call unsubscribe, then the left over messages in the pre-fetch queue are unavailable to the poll call until new messages are fetched from the Broker.

So to conclude. There is absolutely zero point in step 1 and 2 before step 3, as step 2 will always yield nothing even if there are messages in the pre-fetch queue:

1. Unsubscribe so that we can do step 2
2. Consumer - poll so that we can process left over messages in the pre-fetch queue
3. rd_kafka_consumer_close

[edenhill]

Calling Unsubscribe() (or actually Unassign(), which is called when the assignment is revoked) will purge all messages the in the pre-fetch queue.

[agis]

AFAICT this is fixed on librdkafka 0.11.5 (https://github.com/edenhill/librdkafka/commit/26418ae7351e7a7c7f54aac97addcafcbdaebadd)?

[agis]

@edenhill ping! ^

[Knight-Wu]

@edenhill hi, i also encountered this problem, i use this branch: "github.com/confluentinc/confluent-kafka-go-dev/kafka"
, so which confluent-kafka-go version is fixed this issue and can also self static build ?

[agis]

@Knight-Wu which librdkafka version are you using? See https://github.com/confluentinc/confluent-kafka-go/issues/189#issuecomment-415704664.

[Knight-Wu]

@agis I guess this ?

where can i get librdkafka version ?

[edenhill]

kafka.LibraryVersion()

[Knight-Wu]

@edenhill @agis the version is

[edenhill]

@Knight-Wu Can you share relevant parts of your code?

[Knight-Wu]

@edenhill I think is our code problem, thank you very much.