smjn commented on PR #22601:
URL: https://github.com/apache/kafka/pull/22601#issuecomment-4728594656
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]
