[FEATURE] Add KopEventManager to repair metadata consistency of cluster and consistency of consumer group status

[wenbingshen]

Fixes #617

Motivation

When investigating #617, I found the root cause of TimeOut exception and incorrect consumer group status.

The first question is a new problem I found while investigating the cause
1. For example, when the cluster contains two kop nodes, broker1 is the owner broker of all topics in the current cluster, and broker2 is not the leader of any partition.

• 1.1. When the client describesGroup, it needs to connect to the configured brokerList first.
• 1.2. Suppose it connects to broker2 to obtain metadata. For the current server-side metadata processing logic, if the leader nodes of all partitions are not broker2, they are all broker1. Since metadata acquisition proceeds normally, the metadata returned by the brokers node only contains broker1.
• 1.3. So next, the client will send FindCoordinatorRequest to broker1. Broker1 calculates based on the consumer group hash and finds that the coordinator is on broker2.
• 1.4. Because the cluster attribute in the client metadata only has the broker1 node, the client can no longer send DescribeGroupRequest to the broker2, and keep sending the update metadata request to broker1 until it times out.

The second question is described in #617
2. The current GroupCoordinator will not remove the deleted partition from the GroupMetadata after the topic is deleted

Modifications

Add a single-threaded event processing queue on the kop server. It will obtain the current cluster node metadata once at startup and place it locally, and will also register to monitor the /loadbalance/broker subnode and /kop/delete_topics subnode, /loadbalance /broker is used to update the metadata of the kop node list, and /kop/delete_topics is used to update the GroupMetadata which had these deleted topics, and try to change the state of the consumer group.

[wenbingshen]

1. When a new node started, the kop brokers node metadata update is triggered.
   

[wenbingshen]

2. Remove the deleted partition from the GroupMetadata and try to update the consumer group status.

[wenbingshen]

@BewareMyPower There are still some bugs in the code that need to be fixed and related tests added. Before that, can you first take a look at the solution described by this pr? thanks.

[wenbingshen]

Processing cluster brokers metadata updates and processing deleted topics for consumer groups

[wenbingshen]

@BewareMyPower My PR and I are ready. Zk client has been replaced with MetadataStore. PTAL :)

[BewareMyPower]

I've left some comments, PTAL. The main problem is that we should avoid waiting a future in any callback, which was very dangerous and hard to investigate the reason if deadlock happened. There was a similar problem before, see #556.

[wenbingshen]

I've left some comments, PTAL. The main problem is that we should avoid waiting a future in any callback, which was very dangerous and hard to investigate the reason if deadlock happened. There was a similar problem before, see #556.

Thank you for your meaningful comments. Based on your comment, I fixed the relevant code, and for the test, I added timeout=6000 ms. In my local area, the actual test can be completed in just a few hundred milliseconds. It took 1 minute before because I got the judgment condition wrong. I mistakenly wrote describeGroup.get(groupId).state().name().equals("DEAD") as describeGroup.get(groupId).state( ).name().equals("Dead")). PTAL :)

[wenbingshen]

A single test only took around 190 milliseconds.

[wenbingshen]

The failed test does not seem to be related to the PR.

[BewareMyPower]

Overall LGTM, just left a few comments, PTAL.

[wenbingshen]

@BewareMyPower I have addressed your comments. PTAL.