smjn opened a new pull request, #22601:
URL: https://github.com/apache/kafka/pull/22601
* In this PR, we have extracted the record fetch logic from
`ShareGroupDLQStateManager` and encapsulated it in `ShareGroupDLQRecordFetcher`.
* The fetcher also gets the ability to fetch the records from the remote
tier and then returns the combined record map via its public `fetch` method.
* Since the remote record fetch is inherently async, there is a change in
the way the records are populated:
* Earlier records population was lazy in that it was invoked when handlers
were coalesced. Now, we will issue the record fetch call when the handler is
enqueued.
* Furthermore, we also require to additional state in the fetcher so that
local and remote records could be stored before being combined. This is also
divergent from previous code.
* Lastly, we have added tests to `ShareGroupDLQRecordFetcherTest` and
`ReplicaManagerLogReaderTest`.
The record fetcher pseudo code is:
```
state:
endOffset = param.lastOffset
recordMap = {} // offset -> Record (sparse)
result = new CompletableFuture()
fetch() -> Future<Map>: // public
try: runFrom(param.firstOffset)
catch: result.complete({}) // empty on err
return result
runFrom(offset): // synchronous loop
while offset <= endOffset:
adv = fetchLocal(offset)
if adv.isEmpty: return
offset = adv.get
complete()
fetchLocal(readFrom) -> OptionalLong: // empty = stop, present = next
offset
res = logReader.read(readFrom)
if res == null or res.error: result.complete({}); return empty //
abort
if res.isTiered: return fetchRemote(res.descriptor, readFrom) //
delegate
return advanceOrStop(collect(res.records, readFrom), readFrom)
fetchRemote(descriptor, readFrom) -> OptionalLong:
f = logReader.readRemote(descriptor) // async
if not f.isDone:
f.whenComplete((data, ex) -> resumeRemote(readFrom, data, ex))
return empty // resume from
callback
return advanceOrStop(process(readFrom, f.getNow, f.error), readFrom)
resumeRemote(readFrom, data, ex): // runs on reader thread,
after runFrom returned
try:
adv = process(readFrom, data, ex)
if adv <= readFrom: complete()
else: runFrom(adv) // resume; fresh stack
catch: result.complete({})
process(readFrom, data, ex) -> long:
records = (ex != null or data == null) ? EMPTY : data.records //
skip on failure
return collect(records, readFrom)
collect(records, readFrom) -> long: // fill recordMap, return next
offset
next = readFrom
for r in records:
if r.offset < readFrom: skip
if r.offset > endOffset: break
recordMap[r.offset] = r
next = max(next, r.offset + 1)
return next
advanceOrStop(advanced, readFrom) -> OptionalLong: // inlined in real
code
if advanced <= readFrom: complete(); return empty // no progress
return of(advanced)
complete():
result.complete(immutableCopy(recordMap)) //missing
offsets => absent
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
