feat: Reset Consumer upon out-of-band seek

[tmdiep]

Resets the Kafka Consumer state to handle admin seeks pushed from the server.

• SubscriberState was refactored out into SinglePartitionSubscriber. The main motivation was to give each subscriber state its own mutex and update its internal state (for auto-commits) atomically when pulling messages.
  • SingleSubscriptionConsumerImplTest was mostly left as-is to verify that this is a no-op.
• Added SinglePartitionSubscriber::onSubscriberReset() to reset subscriber state and wait for commits when the RESET signal is received from the server.
  • Undelivered messages are discarded. Further auto-commits are prevented until post-seek messages are received.

[dpcollins-google]

I think the majority of your code is correct for the PR. The main thing I'm getting stuck on is what the seek reset behavior should be in the presence of client-initiated seeks. I think its reasonable to say that this is unsupported with a comment in the section here https://github.com/googleapis/java-pubsublite-kafka#about-pubsub-lite-kafka-shim and wherever the equivalent section is in the docs.

[tmdiep]

I think the majority of your code is correct for the PR. The main thing I'm getting stuck on is what the seek reset behavior should be in the presence of client-initiated seeks. I think its reasonable to say that this is unsupported with a comment in the section here https://github.com/googleapis/java-pubsublite-kafka#about-pubsub-lite-kafka-shim and wherever the equivalent section is in the docs.

I'll note in the Devsite docs for admin seek that mixing use of client-initiated and admin seeks is not recommended, as they would clobber each other.

Also, I amended this PR to disable handling admin seeks when the Consumer auto-commit setting is false, since this could also lead to races:

1. Client responds to reset signal. Server moves admin seek cursor to commit cursor table.
2. User-initiated commit overwrites commit cursor.
3. Subscribe stream breaks before sending any messages.
4. Subscribe stream will reconnect reading from the user commit cursor, not the admin seek cursor.