KAFKA-16754: Implemented release acquired records functionality to SharePartition #16430
Conversation
adixitconfluent
force-pushed
the
kafka-16754
branch
from
55be16d
to
261579e
Compare
adixitconfluent
changed the title
…lity along with acquisition lock timeout
adixitconfluent
force-pushed
the
kafka-16754
branch
from
fe9c711
to
6e337ea
Compare
… along with acquisition lock timeout
| } finally { | ||
| lock.writeLock().unlock(); | ||
| } | ||
| return CompletableFuture.completedFuture(Optional.ofNullable(throwable)); |
There was a problem hiding this comment.
The CompletableFuture<Optional<Throwable>> is a bit unusual. Generally, a future with a Throwable would complete exceptionally, rather than completing with a result which is an optional Throwable. Then, instantiating an exception is an expensive operation which captures a stack trace and so on. So, I wonder whether this is really the best pattern for passing back the results from these methods.
There was a problem hiding this comment.
Hmmm, good question.
There are 3 options:
CompletableFuture<Throwable>CompletableFuture<Optional<Throwable>>CompletableFuture<Void>
Option 1 doesn't seems right as it specifies that a Throwable should be returned when completed. I was mostly thinking about option 2 and 3, Option 2 explicitly says that we might have a Throwable which defines the success of the method. Option 3 is more aligned with your suggestion, that future will be completed exceptionally if error occurs else the method success completion shall not result anything.
I went ahead with explicit declaration, also with Option 3, the actual exception gets wrapped in ExecutionException which need wrap/unwrap. Though it's minimal and can be debated when wrapped in Optional. I am fine with either though. Wdyt?
There was a problem hiding this comment.
Since, we need to use the exception to return to the API that functions acknowledge and releaseAcquiredRecords return, I think we can go ahead with option 2. The tradeoffs remain the same with both option 2 and option 3 as mentioned by @apoorvmittal10 .
There was a problem hiding this comment.
Thanks for the PR, LGTM. Just a minor comment and let's follow up on @AndrewJSchofield's suggestion.
| if (inFlightBatch.offsetState() != null) { | ||
| Optional<Throwable> releaseAcquiredRecordsThrowable = releaseAcquiredRecordsForPerOffsetBatch(memberId, inFlightBatch, recordState, updatedStates, stateBatches); | ||
| if (releaseAcquiredRecordsThrowable.isPresent()) { | ||
| throwable = releaseAcquiredRecordsThrowable.get(); |
There was a problem hiding this comment.
Why to fetch the throwable and again wrapping in Optional? The method signature returns CompletableFuture<Optional<Throwable>> itself. Just noticed we need to correct other methods as well, hence I will take that in separate PR while correcting other API methods as well.
About
Implemented release acquired records functionality in SharePartition. This functionality is used when a share session gets closed, hence all the acquired records should either move to
AVAILABLEorARCHIVEDstate. Implemented the following functions -releaseAcquiredRecords- This function is executed when the acquisition lock timeout is reached. The function releases the acquired records.releaseAcquiredRecordsForCompleteBatch- Function which releases acquired records maintained at a batch level.releaseAcquiredRecordsForPerOffsetBatch- Function which releases acquired records maintained at an offset level.Testing
Added unit tests to cover the new functionality added.