Feature: Commit arbitrary offset

[ianwsperber]

Addresses #122 by allowing users to provide offsets to the commitOffsetsIfNecessary method in the eachBatch callback.

Basically the existing API has no interface for committing an arbitrary offset outside a transaction. The recent transactional work in #232 adds an API for transactions to send offsets as participants in a transaction, but that doesn't help if the user wants to commit an offset without opening a transaction.

I'm open to hearing alternatives for exposing this functionality. This seemed the least disruptive for the existing API.

[Nevon]

The implementation looks fine to me. It does feel a little weird to overload commitOffsetsIfNecessary in this way, but I don't really have a better solution. Exposing commitOffsets (with the ability to provide arbitrary offsets) I feel would worsen the ergonomics, since you'd have to know a lot about the internals to understand why it doesn't respect your commit thresholds.

[Nevon]

Since we only care about commitOffsetsIfNecessary in this test, I would suggest changing this to expect it to have been called with expect.objectContaining({ commitOffsetsIfNecessary: expect.any(Function) }), so that this won't start failing just because we added something to the signature of eachBatch

[ianwsperber]

Fair point, I'll change :)

[ianwsperber]

@Nevon Since the user only knows about the behavior of the exposed commitOffsetsIfNecessary method I don't think the overloading will be too weird for them. Originally I'd modified commitOffsetsIfNecessary to accept an optional offsets object, meaning no overload would be necessary, but I thought that was more confusing.

I do think there is still an argument for exposing a commitOffsets method directly on the consumer. I'll actually be using this to commit offsets outside the eachBatch loop, so I'll have to do some juggling to get access to the commitOffsetsIfNecessary method. However IMO it's better to do this first then assess whether there's a need for such a method.