Open indrekj opened 1 week ago
The LastOffset in the flatten_batches function seems to be the correct one. Fetch function even has it as NewBeginOffset0 +1. But the fetch does not return it. If it would, then we wouldn't need to do the https://github.com/kafka4beam/brod/blob/636e448f7d0183dc54fef31776cbe34d5d6f243b/src/brod_utils.erl#L644 call.
We could improve the brod_utils:fetch/4, but it's an exported function and is also used by brod:fetch. We could change the latter to keep backward compatibility, but I don't know if brod_utils is also considered a public interface from the compatibility point of view. @zmstone, what do you think?
Got something working, hopefully it makes sense: https://github.com/kafka4beam/brod/pull/589
A failing test:
Output:
This is the line that throws an exception: https://github.com/kafka4beam/brod/blob/636e448f7d0183dc54fef31776cbe34d5d6f243b/src/brod_utils.erl#L644
The
brod_utils:fetch
(flatten_batches) seems to be control message aware, but at some point something gets lost.I dug pretty deep, but I'm unsure how to fix this.