[server] Increased Wait After Unsubscribe During State Transitions

[KaiSernLim]

Summary

Problem

The problem is that StoreIngestionTask does not always wait for all inflight messages to be processed before transitioning the leader-follower state in the PartitionConsumptionState.

waitAfterUnsubscribe() waits up to 10 seconds for the consumers next poll() (which would indicate that the inflight messages from the last poll() were processed). This can lead to state mismatches such as from the leader-follower transition and follower-leader transition. The 10s timeout has been hit 150K times in the past month.

Mitigation

Several possible solutions were discussed but they all could be complicated. As an immediate action, we can increase the timeout value so that the consumer will more frequently safely unsubscribe instead of timing out. According to the values of the metric consumer_records_producing_to_write_buffer_latency, increasing the timeout to 1800s should cover almost 100% of all cases.

Changes

• Added server config server.wait.after.unsubscribe.timeout.ms to turn the timeout wait in waitAfterUnsubscribe() into a configurable setting, and also increased the timeout:
  • Increased the default value of this timeout from 10s to 1800s / 30m based on the consumer_records_producing_to_write_buffer_latency metric.
  • During shutdown / termination scenarios when KafkaConsumerService#unsubscribeAll() is called, the timeout will remain 10s in order to not block shutdown. If the server config is lower than 10s, then that value will be used instead.

How was this PR tested?

GHCI

Does this PR introduce any user-facing changes?

• No. You can skip the rest of this section.

[KaiSernLim]

Closing to consider approach in #1251 instead.

[gaojieliu]

From my understanding, we don't need to change the constructor of KafkaConsumerService or SharedKafkaConsumer, but just add one param: waitingTimeMS to unsubscribe function call, and when the SIT calls unsubscribe, it knows when to use a longer timeout and when to use a shorter timeout.
By default, it should use a short timeout, and only for leader <-> follower transition, SIT needs to pass a long timeout to unsubscribe call.

Let me know if I have any misunderstanding.