KAFKA-7109: Close cached fetch sessions in the broker on consumer close by stanislavkozlovski · Pull Request #5407 · apache/kafka

stanislavkozlovski · 2018-07-20T17:44:56Z

Previously, the consumer's incremental fetch sessions would time out once the consumer was gone.

stanislavkozlovski · 2018-07-20T17:45:35Z

@cmccabe please review. No easy way to test this as far as I'm aware, integration tests should at least verify it doesn't break anything.

ijuma · 2018-07-25T06:37:21Z

cc @cmccabe

cmccabe · 2018-07-25T16:33:07Z

We should also clear the sessionHandlers map here, just for tidiness.

cmccabe · 2018-07-25T16:34:12Z

Thanks for this, @stanislavkozlovski. Looks good.

We need a test, though. Take a look at clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java You should be able to write something that will verify that closing the Fetcher does the expected thing.

stanislavkozlovski · 2018-07-25T18:25:31Z

Initially I couldn't find a way to test this in FetcherTest.java but after your nudge I looked into it harder and found a way. Thanks @cmccabe.

cmccabe · 2018-07-27T16:16:44Z

Should we also check that there are no partitions contained in the request?

Also we should match the fetch session with the earlier one

Absolutely. Done

cmccabe · 2018-07-28T00:15:05Z

LGTM. Thanks, @stanislavkozlovski.

stanislavkozlovski · 2018-07-30T16:42:58Z

Retest this please

stanislavkozlovski · 2018-08-02T22:13:11Z

Retest this please

ijuma · 2018-08-08T07:00:54Z

Don't we have to do something with the futures returned by send? How do we know if the requests completed? Also, cc @hachikuji.

We had discussed offline that there is no need to wait and ensure the requests completed. It makes sense to make it best effort, otherwise it can slow down client termination and I assume people prefer the client terminates quicker rather than make sure no unused sessions are in the broker

We should always document things like that so that people reading the code are aware of it. My point was more subtle though, how do we make sure we actually sent the request?

More specifically, there's no point in having this if we close the network client before we even get to the point where the request has reached the network layer.

Oh, I did not know that was a possibility. I'll dive in.
Fair point with the documentation

It probably makes sense to wait a short time, but not very long, to send the request to the broker. I'm not sure how that interacts with the rest of the client close logic

Yeah, good catch @ijuma. This is the code for closing the KafkaConsumer:

ClientUtils.closeQuietly(fetcher, "fetcher", firstException); ClientUtils.closeQuietly(interceptors, "consumer interceptors", firstException); ClientUtils.closeQuietly(metrics, "consumer metrics", firstException); ClientUtils.closeQuietly(client, "consumer network client", firstException); ClientUtils.closeQuietly(keyDeserializer, "consumer key deserializer", firstException); ClientUtils.closeQuietly(valueDeserializer, "consumer value deserializer", firstException);

Meaning there's a pretty big chance the client is closed before the request are sent.

But then I read the Javadoc of #send()

Send a new request. Note that the request is not actually transmitted on the * network until one of the {@link #poll(long)} variants is invoked. At this * point the request will either be transmitted successfully or will fail. * Use the returned future to obtain the result of the send. Note that there is no * need to check for disconnects explicitly on the {@link ClientResponse} object; * instead, the future will be failed with a {@link DisconnectException}.

Meaning this code doesn't transmit any requests since poll is not called at all...

I see two approaches to this

Call poll(Long timeout) with the request.timeout.ms configured timeout after the for loop. This will give all requests a grand total of timeout time to be sent. This call will block for a total maximum of request.timeout.ms (default is 30 seconds which seems high to me for this use case). Maybe a total timeout of N seconds, where N is the number of requests?

Call poll((RequestFuture<?> future, long timeout) individually for each sent close request. This blocks on each call, so it's questionable what the timeout should be (1 second?)

In both cases, we will slow down the closing of the consumer for a bit, as we will have to block as far as I can tell.

As a sidenote - this code would also need to handle exceptions as it currently doesn't. My idea is to only catch & log the error.

@cmccabe @ijuma what do you believe is the proper approach?

The consumer close code needs to block until the fetch sessions are closed. KafkaConsumer has close() and close(long timeout, TimeUnit timeUnit). In the case of the latter one, we don't want to wait longer than the specified time.

stanislavkozlovski · 2018-08-08T08:34:00Z

Retest this please

stanislavkozlovski · 2018-10-11T16:53:14Z

wonder if we should handle cases where this is less than 0

Can we not use the Timer that we pass to the Coordinator?

We should consider using the Timer abstraction in the Fetcher methods too.

Yes, agreed

Although if we use the same Timer we ought to make the coordinator close with a timer of timeoutMs, not the min of timeoutMs and requestTimeoutMs, right?

stanislavkozlovski · 2018-10-11T16:53:54Z

@cmccabe I've addressed the latest changes and rebased to trunk

ijuma · 2018-10-13T14:44:58Z

Can we not use closeQuietly?

I don't think so since it won't be able to pass the Timer class to fetcher.close. Maybe we could introduce a new interface called CloseableWithTimeout/TimeoutCloseable and have closeQuietly handle that as well?

Or have one that takes a lambda so that the caller can do the close. Similar to what we have for Scala.

ijuma · 2018-10-13T14:45:53Z

Can we not use the Timer that we pass to the Coordinator?

ijuma · 2018-10-13T14:46:57Z

We should consider using the Timer abstraction in the Fetcher methods too.

ijuma · 2018-10-14T16:00:27Z

Why is the Math.min not done any longer?

That's what I asked in #5407 (comment)
Since the Timer will be shared across multiple close() methods, having it as short as one requestTimeoutMs may not give sufficient time for fetcher.close(). Do you think it's okay to use a shared timer with the minimum for both close calls?

We could update the timer so that the min was the requestTimeoutMs for the second close too.

In the end I think the timer for the second close should be the minimum of what time is left out of timeoutMs and requestTimeoutMs.

stanislavkozlovski · 2018-11-15T14:25:34Z

Retest this please

stanislavkozlovski · 2018-12-03T12:59:52Z

Retest this please

Previously, the consumer's incremental fetch sessions would time out once the consumer was gone.

ijuma · 2019-08-26T13:17:49Z

It may be worth reviving this, it requires a rebase.

showuon · 2022-02-04T07:57:34Z

@stanislavkozlovski , encountered similar errors and I agree we should close the incremental fetch sessions when consumer closed. Are you able to complete this PR? If not, I can co-author with you to help complete this PR to merge this good improvement. Thank you.

showuon · 2022-02-08T10:22:49Z

@stanislavkozlovski , I'm going to take over this ticket tomorrow if you don't mind. Thank you.

stanislavkozlovski · 2022-02-08T12:45:23Z

@showuon yes, please feel free. apologies for being slow to respond and missing the previous messages about reviving this. I will go ahead and close this PR now

showuon · 2022-02-08T12:58:39Z

Thank you!

ijuma reviewed Jul 21, 2018

View reviewed changes

Comment thread clients/src/main/java/org/apache/kafka/clients/FetchSessionHandler.java Outdated

cmccabe reviewed Jul 25, 2018

View reviewed changes

cmccabe reviewed Jul 27, 2018

View reviewed changes

ijuma reviewed Aug 8, 2018

View reviewed changes

stanislavkozlovski commented Oct 11, 2018

View reviewed changes

ijuma reviewed Oct 13, 2018

View reviewed changes

ijuma reviewed Oct 14, 2018

View reviewed changes

stanislavkozlovski added 3 commits November 26, 2018 13:33

kafkatest: Match and compare *-SNAPSHOT versions in KafkaVersionTest

f8df762

Merge branch 'trunk' of https://github.com/apache/kafka into trunk

bdbd646

Merge branch 'trunk' of https://github.com/apache/kafka into trunk

ca375de

stanislavkozlovski added 5 commits December 3, 2018 13:04

KAFKA-7109: Close cached fetch sessions in the broker on consumer close

86baa71

Previously, the consumer's incremental fetch sessions would time out once the consumer was gone.

KAFKA-7109: Add test for closing fetch sessions in broker

ee577ba

KAFKA-7109: Improve fetcher session handler close test

dfa222f

KAFKA-7109: Add timeout for closing fetching sessions

94417ac

Use Timer class for timeouts

dc33ae0

stanislavkozlovski closed this Feb 8, 2022

divijvaidya mentioned this pull request Sep 5, 2022

KAFKA-7109: Close fetch sessions on close of consumer #12590

Merged

Conversation

stanislavkozlovski commented Jul 20, 2018

Uh oh!

stanislavkozlovski commented Jul 20, 2018

Uh oh!

Uh oh!

ijuma commented Jul 25, 2018

Uh oh!

cmccabe Jul 25, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cmccabe commented Jul 25, 2018

Uh oh!

stanislavkozlovski commented Jul 25, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cmccabe commented Jul 28, 2018

Uh oh!

stanislavkozlovski commented Jul 30, 2018

Uh oh!

stanislavkozlovski commented Aug 2, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

stanislavkozlovski commented Aug 8, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

stanislavkozlovski commented Oct 11, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

stanislavkozlovski commented Nov 15, 2018

Uh oh!

cmccabe Jul 25, 2018 •

edited

Loading