changefeedccl: better handle kafka quotas

[amruss]

Kafka users can set a data quota in kafka. As a high throughput, high scale database, we can easily run up on those quotas. Customers cannot have an arbitrarily large kafka cluster to handle traffic bursts.

Likely what we want to do is:

• exponentially backoff when we see a quota exceeded error from kafka (for the length of time kafka tells us to?)
• implement pushback in this case ^

We should also add end-to-end testing for the scenario where kafka is purposely slow

Jira issue: CRDB-21692

Epic CRDB-21691

[blathers-crl]

cc @cockroachdb/cdc

[blathers-crl]

cc @cockroachdb/cdc

[shermanCRL]

Overall, the answer seems be first to design a throttling system for changefeeds, similar in spirit to memory monitoring / admission control / quota pools.

Consider scope per pool/bucket: is it per-cluster, per-database or per-changefeed? Actually, perhaps changefeeds could be grouped into arbitrary pools.

[miretskiy]

@HonoreDB , I believe you're working on this right now.

[miretskiy]

Not going to do this in 23.2.
Assigning to myself to attempt to create an upstream sarama patch to add the capability required for quota pushback implementation.

[wenyihu6]

Creating a list of sub-issues to track the work that still needs to be done:

• Test and upgrade sarama version (make sure no regression). (cdc: test and upgrade sarama version  #117522)
• Add observability by injecting sarama metrics interceptor. (cdc: add observability metrics to throttling behaviour in kafka #117618)
• Add client quota push back -> likely token bucket based using DescribeClientQuotas. (changefeedccl: explore whether we can gauge kafka quota usage #92759)

[wenyihu6]

I discussed this with yev and my understanding of this issue is:

• The newer version of sarama acknowledges the throttling behaviour from kafka and starts throttling upon receiving the response. https://github.com/IBM/sarama/commit/5ac5dc0e90b2317118b193cf1aa03dc5f2122539/
• My understanding was that when sarama starts throttling after receiving the kafka response, this effect would propagate along the changefeed pipeline and cause rangefeed to slow down and emit less messages as well until the throttling is done. So no work would be needed on our client side to react to this throttling effect since sarama's effect gets propagated.

The work needs to be done in this area is:

1. Upgrade sarama version - done deps: upgrade Shopify/sarama v1.38.1 to IBM/sarama v1.42.1 #117544
2. Add observability metrics into sarama so that we know when sarama starts throttling and we can tell customer why their changefeed is slow and they would need to adjust their kafka quota configuration accordingly changefeedccl: add observability metrics into sarama code #117693
3. Add per-changefeed configuration so that user can specify different kafka quota configuration for different changefeeds changefeedccl: allow per changefeed kafka quota config #118643. This is done by changing the sarama.clientID. Users can specify different quota based on the client ID according to https://kafka.apache.org/documentation/#quotas. I'm still doing more testing on this PR.

@miretskiy Could you help me confirm my understanding here?

cc:@rharding6373

[miretskiy]

Your understanding is exactly right.
It is possible that the above solution might not be enough (I still worry about the possibility
of changefeeds constantly getting pushback signal; respecting this signal; and then restarting
with a blast of messages -- which would cause more pushback)... However, I think this gives
a good first step, and it's not clear yet if we'd need to do anything else.

[wenyihu6]

Your understanding is exactly right. It is possible that the above solution might not be enough (I still worry about the possibility of changefeeds constantly getting pushback signal; respecting this signal; and then restarting with a blast of messages -- which would cause more pushback)... However, I think this gives a good first step, and it's not clear yet if we'd need to do anything else.

Thanks for the confirmation!

[wenyihu6]

Your understanding is exactly right. It is possible that the above solution might not be enough (I still worry about the possibility of changefeeds constantly getting pushback signal; respecting this signal; and then restarting with a blast of messages -- which would cause more pushback)... However, I think this gives a good first step, and it's not clear yet if we'd need to do anything else.

For my education, I wanted to check how this effect gets propagated. Are we just busy waiting here https://github.com/cockroachdb/cockroach/blob/edd9e94cfd8489492816617200aae8c12946b83f/pkg/ccl/changefeedccl/sink_kafka.go#L558-L577 since the producer will be blocked while waiting for throttled https://github.com/IBM/sarama/blob/2767191b19b2e190f7095f21cac2a014de80e92c/broker.go#L1009-L1010. And since the sarama is waiting, we also wait in workerLoop() which in turn slows down the emitting messages. If this is correct, could you point to me where emitting less messages in turn slow down rangefeed in the code?

[wenyihu6]

@miretskiy ^ : ) last question 😬

[miretskiy]

Let's discuss offline...