Transition to RUNNING only after offset-reset happens

[ocadaruma]

Summary

• Currently, ProcessorSubscription's state transitions to RUNNING at the end of rebalance listener (onPartitionsAssigned)
• So, offset-reset could be happen in later poll() after transitioned to RUNNING
• Sometimes, this could cause unintuitive behavior to the user.
  • Even for us. We were not aware about this behavior until we found the benchmark was not running (https://github.com/line/decaton/pull/81)

Example

• Some users set auto.offset.reset = none when they migrate their consumer to Decaton from other framework (which likely includes a lot of code changes), to make sure group.id is configured correctly so can continue consumption from committed offset.
  • So they tried to detect the group.id-mistake by checking the subscription transitioned to RUNNING or not. (auto.offset.reset = none throws an exception when starts up if group.id doesn't exist, so they assumed the subscription doesn't transition to RUNNING if group.id is not configured properly)
• But, in fact, subscription transitions to RUNNING even if there's a mistake in group.id. (and transition to SHUTTING_DOWN -> TERMINATED soon)