feat: Ability to reset subscriber upon out of band seek

[tmdiep]

Prepares for handling the RESET signal from the server, by:

• Discarding outstanding acks for delivered messages.
• Waiting for the committer to flush pending commits and receive the acknowledgment from the server.
• Then resetting subscriber state, including canceling any in-flight client seeks.

Notes:

• This PR is effectively a no-op as SinglePartitionSubscriber.onSubscriberReset returns false for now.
• The next PR will fully handle admin seeks by setting InitialSubscribeRequest.initial_location and implement SinglePartitionSubscriber.onSubscriberReset. Omitted to keep the size of the PR down.
• CommitState.complete had to be refactored to just throw (and not abort) because CommitterImpl.waitUntilEmpty should throw when there are excess commit responses, since the committer will shortly shut down.

[dpcollins-google]

The next PR will fully handle admin seeks by setting InitialSubscribeRequest.initial_location. Omitted to keep the size of the PR down.

This leaves the client in an indeterminate state where it will drop a bunch of acks but not actually respect the seek on the reconnect. I don't think this is a good state to be in at head. Can you remove the changes to SubscriberImpl from this PR and move them to the next one, so at least if someone happens to (I don't think they will) pick up this intermediate state it will have the same behavior as the existing code (i.e. not reset?)

[tmdiep]

Can you remove the changes to SubscriberImpl from this PR and move them to the next one, so at least if someone happens to (I don't think they will) pick up this intermediate state it will have the same behavior as the existing code (i.e. not reset?)

Good point. Instead I just changed SinglePartitionSubscriber.onSubscriberReset to return false, which effectively becomes a no-op.

This will keep the size of the next PR down and avoids re-reviewing many files.