Async producer can overflow itself at high rate #805
Comments
|
Ok, it looks like Still, this doesn't seem right. Message size back pressure should be controlled by |
|
|
There's no doubt in that.
Yes, but the constraint of maximum message size gets violated, because no backpressure is applied when message size crosses this boundary. This is exactly the issue. I set |
When not using compression (your ticket specifies
|
|
I notice that you have the following configuration values: Since these values are equivalent they are usually set to the same value. |
|
Just getting into investigating here, but I believe what we're trying to say is that because we have such a high rate of production on one topic-partition, the first statement in the select is always being run and the following lines don't have a chance to run: Since those don't have a chance to run, we're never rolling over the buffer, and it grows too large. @bobrik , please correct me if I'm wrong. Still looking into it more though. I think the fact that setting Also, I'm auditing our producers to ensure that they've all got |
|
If your produce rate is that high, you should still hit a hard block when |
|
Thanks. I'll update our patch to ensure we're hitting that condition and get back. |
|
@eapache what you say sounds good in theory, but it's just now how I see it working in real life. Consider the following patch: diff --git a/vendor/github.com/Shopify/sarama/async_producer.go b/vendor/github.com/Shopify/sarama/async_producer.go
index e7ae8c2..27c2f10 100644
--- a/vendor/github.com/Shopify/sarama/async_producer.go
+++ b/vendor/github.com/Shopify/sarama/async_producer.go
@@ -2,6 +2,7 @@ package sarama
import (
"fmt"
+ "log"
"sync"
"time"
@@ -249,6 +250,7 @@ func (p *asyncProducer) dispatcher() {
}
if msg.byteSize() > p.conf.Producer.MaxMessageBytes {
+ log.Printf("Got message size bigger than allowed max message bytes %d > %d", msg.byteSize(), p.conf.Producer.MaxMessageBytes)
p.returnError(msg, ErrMessageSizeTooLarge)
continue
}
@@ -577,6 +579,8 @@ func (bp *brokerProducer) run() {
var output chan<- *produceSet
Logger.Printf("producer/broker/%d starting up\n", bp.broker.ID())
+ maxMessageSize := 0
+
for {
select {
case msg := <-bp.input:
@@ -585,6 +589,11 @@ func (bp *brokerProducer) run() {
return
}
+ if msg.byteSize() > maxMessageSize {
+ maxMessageSize = msg.byteSize()
+ log.Printf("Biggest message seen so far: %d bytes", maxMessageSize)
+ }
+
if msg.flags&syn == syn {
Logger.Printf("producer/broker/%d state change to [open] on %s/%d\n",
bp.broker.ID(), msg.Topic, msg.Partition)
@@ -610,6 +619,7 @@ func (bp *brokerProducer) run() {
}
if bp.buffer.wouldOverflow(msg) {
+ log.Println("!!! OVERFLOW !!!")
if err := bp.waitForSpace(msg); err != nil {
bp.parent.retryMessage(msg, err)
continue
@@ -627,6 +637,7 @@ func (bp *brokerProducer) run() {
case <-bp.timer:
bp.timerFired = true
case output <- bp.buffer:
+ log.Printf("Roll over, bufer size: %d", bp.buffer.bufferBytes)
bp.rollOver()
case response := <-bp.responses:
bp.handleResponse(response)
@@ -634,6 +645,7 @@ func (bp *brokerProducer) run() {
if bp.timerFired || bp.buffer.readyToFlush() {
output = bp.output
+
} else {
output = nil
}
diff --git a/vendor/github.com/Shopify/sarama/produce_set.go b/vendor/github.com/Shopify/sarama/produce_set.go
index 9fe5f79..4886032 100644
--- a/vendor/github.com/Shopify/sarama/produce_set.go
+++ b/vendor/github.com/Shopify/sarama/produce_set.go
@@ -147,6 +147,7 @@ func (ps *produceSet) readyToFlush() bool {
return true
// If we've passed the byte trigger-point
case ps.parent.conf.Producer.Flush.Bytes > 0 && ps.bufferBytes >= ps.parent.conf.Producer.Flush.Bytes:
+ // log.Printf("ready to flush because buffer bytes are big enough: %d >= %d", ps.bufferBytes, ps.parent.conf.Producer.Flush.Bytes)
return true
default:
return false
The config looks like this: conf.Producer.MaxMessageBytes = 33554432
conf.Producer.Flush.Bytes = 16 << 20Let's look at the output: So my bugger message was around 20kb, but Kafka still complains about too big messages. Limiting @abraithwaite |
|
Can you provide the config of the kafka broker please? |
|
Also, to verify, this is all running still with |
|
Still running with Broker config: |
|
Are any of those config values overridden for the specific topic you're using (https://kafka.apache.org/documentation/#topic-config)? |
|
@eapache we don't have any topic level overrides: |
|
I'm going to "think aloud". A few things don't line up:
1 is inconsistent with 2, since the broker should only return that error when the message we're sending is larger than the configured maximum. 2 is inconsistent with 3, since the maximum request size and the maximum message size have nothing to do with each other. Also worth noting: there's nothing unusual about the config here, and while the message rate is high, Sarama has been benchmarked and stress-tested fairly thoroughly without anybody else reporting this issue. At this point I can't even generate a consistent hypothesis which explains all of the facts. So let's get some more data and see if something pops:
|
|
Also worth noting in the interest of completeness: the SHA you're running is a fairly old version at this point, but I don't think there have been any changes since that version which would affect this particular issue. Upgrading might be something to try; if nothing else it would give us another data point. |
|
And in the interest of "more data won't hurt", could you provide the standard Sarama logs for one of these issues as well? You've provided some output of your custom logging statements, but the defaults (the ones going to |
|
We'll work on getting you some more information on this, but I do have one clarification question to ask regarding how the flush timing is handled: If we check readyToFlush here: And we flush here: The Seems like part of the reason we're seeing issues is because the Gonna investigate how the other clients handle this though. Edit: seems like the java producer sends immediately upon the threshold crossing for |
|
This is an experimental cluster and I tested compression on brokers at some point. I enabled debug logs on one broker with compression enabled and here's what I found: Disabling |
|
I guess we have this: https://issues.apache.org/jira/browse/KAFKA-1718 |
The specific scenario in question is when:
Right now we just accept the message, since we can safely handle it. We could in principle start applying backpressure at this point, but all that would accomplish would be to artificially limit throughput, especially since the flush-trigger and the maximum-supported are usually substantially different.
Except for
If they behave this way, Honestly, those two config values ( |
Yes, you have a new variant of that issue which seems to be caused by the broker doing compression even when the client sends uncompressed messages. Since the client is not compressing the messages at all, it has no way to even know to work around the issue. It may be worth filing another upstream ticket for the new bug and referencing the old one, since the new one is more problematic and harder to magically work around for the client. For such high-throughput scenarios you probably want to enable client-side compression anyway, but if that doesn't work for you then setting |
|
Closing as this is an upstream bug. I'm happy to keep discussing flush logic though if there are still questions around that. |
|
Thanks for the thoughtful response. Although I have one quick follow up question: How can I identify whether a produce request is in-flight in the code we've highlighted? I'm assuming that It'd be nice to try (nonblocking) As @bobrik mentioned, we found our issue after all so I guess this can be closed. Thanks for all your help! Edit: you beat me to it. |
Yup. There is a tiny bridge goroutine (https://github.com/Shopify/sarama/blob/0fb560e5f7fbcaee2f75e3c34174320709f69944/async_producer.go#L542-L553) which ranges over
This isn't needed, and I think stems from an earlier misunderstanding. You wrote:
Which is simply not true. Per https://golang.org/ref/spec#Select_statements if multiple cases are ready one will be chosen at random. The fact that the input case is at the top will not starve the output in high-throughput scenarios. |
|
Ah, that's what I knew the behavior was, but then when I went to the docs:
And then stopped reading, confusing myself. Looks like I need to work on my RTFM skills a bit more again. Thanks again for all the help. |
Versions
Please specify real version numbers or git SHAs, not just "Latest" since that changes fairly regularly.
Sarama Version: b1da175
Kafka Version: Irrelevant
Go Version: b1da175
Configuration
What configuration values are you using for Sarama and Kafka?
Sarama async producer:
Logs
Sarama logs:
Problem Description
Problem is in this function:
We produce messages at such rate (100K/s+) so
selectalways picks up processing of new incoming messages over rolling over and flushing existing ones.I applied to following patch:
The output looks like this: https://gist.github.com/bobrik/27071d61d5ec98ed15ffd1cb5331f3f4.
Without
log.FatalI've seen buffer sizes go as high as 40MB, which is bigger thanmessage.max.bytes=33554432in our cluster.The solution is probably to do
rollOveroutside ofselect.The text was updated successfully, but these errors were encountered: