Fix to shutdown PrefetchRecordsPublisher in gracefull manner #857
Conversation
Previously when the lease expires PrefetchRecordsPublisher shutdown the process forecefully by interupting the threads, which lead to leak in apache http client connection Now changed to code to shutdown the PrefetchRecordsPublisher process in more gracefull manager and handled interrupted exception
madagascar22
force-pushed
the
graceful_prefetcher_shutdown
branch
from
4b3bc92 to
b72b9b0
Compare
| @@ -78,6 +78,9 @@ | |||
| @KinesisClientInternalApi | |||
| public class PrefetchRecordsPublisher implements RecordsPublisher { | |||
| private static final String EXPIRED_ITERATOR_METRIC = "ExpiredIterator"; | |||
| // Since this package is being used by all KCL clients keeping the upper threshold of 60 seconds | |||
| private static final Duration AWAIT_TERMINATION_TIMEOUT = Duration.ofSeconds(60); | |||
There was a problem hiding this comment.
How did we get this value? Was there tests run?
There was a problem hiding this comment.
Keeping this as upper threshold value for a prefetcher to end gracefully or else we will forcefully terminate.
…isher Since clients can configure there own awaitTerminationTimeoutMillis, add it as sepearate parameter with default value
| } | ||
| } catch (InterruptedException e) { | ||
| // Preserve interrupt status | ||
| Thread.currentThread().interrupt(); |
There was a problem hiding this comment.
Shouldn't we set the interrupt flag after shutting down the thread-pool in the following line?
There was a problem hiding this comment.
Oh yes, we need to shutdown and then set the interrupt or else shutdown will fail. I will update the same.
| try { | ||
| resetLock.readLock().lock(); | ||
| makeRetrievalAttempt(); | ||
| } catch(PositionResetException pre) { | ||
| log.debug("{} : Position was reset while attempting to add item to queue.", streamAndShardId); | ||
| } catch (Throwable e) { |
There was a problem hiding this comment.
I'm not sure why we wrote this exception handling this way originally, but I feel like making this layer InterruptedException and making the final layer Throwable would be a bit more elegant
There was a problem hiding this comment.
Since we are going to log the error message regardless of the InterruptedException and Throwable. Its good to handle InterruptedException inside the Throwable catch block or else we may simply create a separate function for log message.
|
LGTM. Build is failing for flaky reasons again. Approving. |
|
hi, side effect of this change is, KCL is throwing this error: |
|
This is expected graceful shutdown scenario because this fix allows to make the intentional shutdown decision instead of interrupting the thread using shutdownNow(). |
|
if this is expected behaviour why is it logging an error and throwing exception? surely it should let the worker drain the queue too? perhaps it should look at |
Description of changes:
Previously when the lease expires PrefetchRecordsPublisher shutdown
the process forecefully by interupting the threads,
which lead to leak in apache http client connection
Now changed to code to shutdown the PrefetchRecordsPublisher
process in more gracefull manager and handled interrupted exception
By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.