Add gRPC stream API gateway implementation

[npepinpe]

Description

When a client sends a job stream call (as defined in #11708), the gateway should keep track of these in an in-memory register.

This will be a long living stream observer. It's important that the gateway properly keep track of it, including when it is closed (gracefully or not, including cancellations), and should only be available to receive jobs when it is ready.

See the prototype for an example.

Clients which are logically equivalent based on their activation properties should be grouped together using some unique ID. This unique ID will be used to communicate with the broker when it comes to adding/removing workers, and pushing jobs.

When a new logical client is added, then gateway should send a AddWorker request to all brokers. If the broker returns that the request is invalid, then the error should be propagated to the client. Any other errors should be retried.

Note We may want to limit this of course, but in theory if a broker never replies successfully but the request is correct, likely the broker is dead and will be removed from the topology soon.

If there are no brokers, the client is not cancelled or closed.

When a new broker is added to the topology, an AddWorker request should be sent for all workers.

Note We can look into batching this as an optimization in a further iteration.

When a broker is removed from the topology, nothing is to be done.

When the gateway is shutting down, it should send a RemoveWorkers with its own member ID to all brokers. Requests are not retried here, of course, it's just a best-of effort to be polite.

[deepthidevaki]

A first version for stream-management support for gateway is available in PR #12116. For easy tracking, I'm listing pending tasks here:

• Aggregate multiple client streams with same metadata and streamType to a single stream. This should be done in transport, and transparently to the consumers of the API ClientStreamer#add(streamType, metadata, consumer). Aggregate multiple client streams with same metadata and streamType to a single stream #12253
• Re-visit how to handle node restarts. Currently, it fully relies on MembershipService to notify when a node is added or removed. However, there are some edge-cases which we have to handle. For example, if the broker restarts too fast, gateway may not detect the broker is dead, and as a result will not notify when the broker is restarted.
  • This is not a problem. see comment
• Optimal handling of retries. We might have to stop retrying sending add/remove request to a server at some point. Currently, it is retried indefinitely. It might also be good to retry with a backoff.
• Integrate transport stream-management with gateway so that implementation of grpc api Add gRPC job stream API #11708 can use them.

[npepinpe]

Can we add these points to the open questions in the epic issue? We should evaluate the cost of not doing these now, and the benefits of adding them in the second phase. If we're not sure about the benefits, then identify scenarios which will help us evaluate these.

Except the last one of course, that we just have to do 😄

[deepthidevaki]

Can we add these points to the open questions in the epic issue?

I expect that most of these will be done in the scope of this issue. Let's first discuss what is necessary and what can be postponed, and then move them to open questions. I will not close this issue until then. So we won't loose track of it.

[deepthidevaki]

Another edge case to consider: If a client stream is removed, while there is an ongoing "AddRequest" retry, it is possible that the stream is added to the broker after it has been removed.

• Allow cancelling ongoing retries

[deepthidevaki]

• Re-visit how to handle node restarts. Currently, it fully relies on MembershipService to notify when a node is added or removed. However, there are some edge-cases which we have to handle. For example, if the broker restarts too fast, gateway may not detect the broker is dead, and as a result will not notify when the broker is restarted.

This is not a problem. When a node is restarted, it's incarnation number will be chaged. Swim detects this and post and MEMBER_REMOVED, and MEMBER_ADDED event. So we will always get notified of node restarts, and we will re-send AddStream requests.

https://github.com/camunda/zeebe/blob/c055d58fc86afa675e27f7c8b7e6ce6110a29592/atomix/cluster/src/main/java/io/atomix/cluster/protocol/SwimMembershipProtocol.java#L251

    // If the term has been increased, update the member and record a gossip event.
    else if (member.incarnationNumber() > swimMember.getIncarnationNumber()) {
      // If the member's version has changed, remove the old member and add the new member.
      if (!Objects.equals(member.version(), swimMember.version())) {
        members.remove(member.id());
        post(new GroupMembershipEvent(GroupMembershipEvent.Type.MEMBER_REMOVED, swimMember.copy()));
        ....
        LOGGER.debug("{} - Evicted member for new version {}", localMember.id(), swimMember);
        post(new GroupMembershipEvent(GroupMembershipEvent.Type.MEMBER_ADDED, swimMember.copy()));
        recordUpdate(swimMember.copy());
      }