Obtain the remote broker IP address corresponding to the sending message through Sarama

[freemindLi]

Using Sarama to send messages to Kafka cluster asynchronously and receive the returned message from the successes channel, how can I know which blocker received the message and responded to it (yes, I want to know the broker IP address of receiving each message); At present, when sending messages to Kafka cluster, some of the responses are very late, and some of the responses are very small. We need to know which brokers are slow in response or network delay. So we plan to type out the broker IP address of the response in the Success channel to locate the specific broker conveniently. But, we don't see that Sarama has this function. Please tell me how to solve it

[machicao2013]

Please tell me how to solve it

[machicao2013]

我也遇到了类似的问题，请问怎么获取blocker ip？

[machicao2013]

Using Sarama to send messages to Kafka cluster asynchronously and receive the returned message from the successes channel, how can I know which blocker received the message and responded to it (yes, I want to know the blocker IP address of receiving each message); At present, when sending messages to Kafka cluster, some of the responses are very late, and some of the responses are very small. We need to know which blockers are slow in response or network delay. So we plan to type out the blocker IP address of the response in the Success channel to locate the specific blocker conveniently. But we don't see that Sarama has this function. Please tell me how to solve it

请问这个问题解决了没

[freemindLi]

Using Sarama to send messages to Kafka cluster asynchronously and receive the returned message from the successes channel, how can I know which blocker received the message and responded to it (yes, I want to know the blocker IP address of receiving each message); At present, when sending messages to Kafka cluster, some of the responses are very late, and some of the responses are very small. We need to know which blockers are slow in response or network delay. So we plan to type out the blocker IP address of the response in the Success channel to locate the specific blocker conveniently. But we don't see that Sarama has this function. Please tell me how to solve it

请问这个问题解决了没

No,look at the Sarama source code, can't get the remote broker IP from the responses that from the success channel normally. Maybe you can set the timeout time a little bit smaller, so that requests that haven't responded for more than a little time will report an error（get error info from ERROR channel）. In this way, you can see which broker you are interacting with！

[machicao2013]

Using Sarama to send messages to Kafka cluster asynchronously and receive the returned message from the successes channel, how can I know which blocker received the message and responded to it (yes, I want to know the blocker IP address of receiving each message); At present, when sending messages to Kafka cluster, some of the responses are very late, and some of the responses are very small. We need to know which blockers are slow in response or network delay. So we plan to type out the blocker IP address of the response in the Success channel to locate the specific blocker conveniently. But we don't see that Sarama has this function. Please tell me how to solve it

请问这个问题解决了没

No,look at the Sarama source code, can't get the remote broker IP from the responses that from the success channel normally. Maybe you can set the timeout time a little bit smaller, so that requests that haven't responded for more than a little time will report an error（get error info from ERROR channel）. In this way, you can see which broker you are interacting with！

如果是把超时时间调小才能获取到ip，那就没有意义了。我希望获取到broker ip，然后进行上报。

[dnwe]

The way the Kafka protocol works is that messages are always sent (produced) to the current leader for the given topic partition. If you initialise a "client" and then pass that in to your producer (rather than letting it handle its own) then you can access the metadata information from that.

i.e., (pseudo code)

client, err := sarama.NewClient(brokers, cfg)
if err != nil {
    log.Fatalf("error creating producer client: %v", err)
}
defer client.Close()
producer, err := sarama.NewAsyncProducerFromClient(client)
if err != nil {
    log.Fatalf("error creating producer: %v", err)
}
defer producer.Close()
producer.Input() <- &sarama.ProducerMessage{
    Topic:     topic,
    Value:     sarama.StringEncoder(time.Now().String()),
    Headers:   []sarama.RecordHeader{},
    Timestamp: time.Now(),
}
message := <-producer.Successes()
leaderID := -1
if leader, err := client.Leader(message.Topic, message.Partition); err != nil {
    log.Printf("failed to find leader for topicpartition %s-%d", message.Topic, message.Partition)
} else {
    leaderID = leader.ID()
}
log.Printf(
    "Message produced: value = %s, topic = %s-%d, leader = broker %d\n", message.Value, message.Topic, message.Partition, leaderID)

[github-actions[bot]]

Thank you for taking the time to raise this issue. However, it has not had any activity on it in the past 90 days and will be closed in 30 days if no updates occur. Please check if the main branch has already resolved the issue since it was raised. If you believe the issue is still valid and you would like input from the maintainers then please comment to ask for it to be reviewed.

[dnwe]

Closing as assumed to be fixed by the above comment