[EPIC] Generalized command distribution

[korthout]

Description

We know how to generalize the deployment distribution concept to use it for all record value types. The next step is to break this down into technical tasks.

• Can we generalize the distribution logic for different record value types? #11218

Open questions

• how can we deal with reaching the maxMessageSize twice as fast?

For deployment, the proposal requires us to write the deployed resources twice in the same record batch, i.e. once in the Process:Created (or DecisionRequirements:Created) events, and once in the RecordDistribution:Started event. That means that we reach the maxMessageSize twice as fast. In other words, users will no longer be able to deploy some of the resources they could before. The team proposes to solve this by doubling the current maxMessageSize. We will discuss this with the Zeebe Distributed Platform team, as they may challenge this.

ZDP has suggested to:

• introduce a soft limit (MAX_MESSAGE_SIZE) and a hard limit (🤷 arbitrary, perhaps half the segment size) for record batches, where some processors can choose to ignore the soft limit
• increase the limit for all record batches, and instead produce warnings when record batches exceed the MAX_MESSAGE_SIZE

Which it will be is not yet decided.

• Can we treat the distributed command as a user command and acknowledge it as a response?

Although this question is unanswered, we've decided to move forward without using this

This question came up while deciding 'can we de-deprecate a method in the InterPartitionCommandSender?': Instead of processing distributed commands differently by adding a post-commit task to Acknowledge the command, we could treat the distributed command like other user commands and respond to it. Instead of sending the response to the gateway, it would write a command to another partition.

• can we de-deprecate a method in the InterPartitionCommandSender?

Yes, we will de-deprecate the method

Another topic that should be discussed with the Zeebe Distributed Platform (ZDP) team is de-deprecating the method to send a command to another partition where the sending partition determines the record key. This method was deprecated immediately when it was introduced with the new InterPartitionCommandSender API to avoid us from using it for other cases than Deployment Distribution. However, we could not figure out the reason why this method should not be used. Actually, the team sees reasonable use cases where a partition can decide the record key when it sends it to another partition. As the key encodes the partition id it was generated on, it clearly identifies an entity to exist on that specific partition. We should discuss this with the Zeebe Distributed Platform team to understand why this method was deprecated in the first place and push to revise that stance.

After discussing with ZDP, we agreed that it is necessary for Deployments to re-use the key. The deployment must be available on all partitions with the same key. Otherwise, resource deletion would not be possible because we somehow need to refer to a single resource (Deployment). Initially, we thought it must also be available for process creation, but the process key is already part of the ProcessMetadata which is part of the Deployment command that is distributed to the other partitions.

Also part of the decision is that we clearly document the usage of this method. When is it allowed to use it and when not. It must also be clear why the method must exist.

For full details of the discussion, please find the associated (archived) Slack channel #top-zeebe-dedeprecate-send-command-with-key.

Concept and goal

Some entities in Zeebe are cross-cutting partitions. For example, when you deploy a new Process, you want to be able to start Process Instances of that Process on any of the partitions. But the gateway writes the Deployment:Create command only on a single partition when it handles a call to the DeployResource RPC. The engine has to communicate this command to the other partitions (i.e. each partition has its own engine). This communication is unreliable because of the nature of distributed systems. We need a way to communicate commands between the partitions reliably. We call this mechanism command distribution.

Several features require this same mechanism. Instead of building the same logic multiple times, we should generalize the concept to reuse it for all of these. This maintenance task will unblock:

• [EPIC] Support signal events #10777
• [EPIC] Resource definition deletion #9576
• Suspend and Resume Processes/Instances/Tasks
• ... and potentially others

We can achieve this with the following proposed solution:
Introduce a new value type called CommandDistribution, which wraps the command we want to communicate reliably to other partitions.

An initial STARTED intent ensures we store that command in the state (for lookups at retry), and a final FINISHED event can be applied to remove this command from the state.

Then, for every partition, we have intents DISTRIBUTING, ACKKNOWLEDGE, and ACKNOWLEDGED to keep track of the distribution to that specific partition.

On the receiving partition, we write the encapsulated command directly. Its key contains the information that it was distributed (and from which partition), so the engine knows the send the ACKNOWLEDGE back to the distributing partition.

Task breakdown

Tasks

Preview Give feedback

1. Introduce a new RecordDistribution record zeebe#11657
   component/engine kind/toil version:8.2.0 version:8.2.0-alpha5 +4
   
2. Add state and ColumnFamilies for command distributions zeebe#11868
   component/engine kind/feature version:8.2.0 +3
   
3. Apply RecordDistribution events zeebe#11658
   component/engine kind/toil version:8.2.0 +3
   
4. Acknowledge distributed commands zeebe#11659
   component/engine kind/toil version:8.2.0 +3
   
5. Introduce record distribution behavior zeebe#11660
   component/engine kind/toil version:8.2.0 +3
   
6. Version event appliers zeebe#12764
   component/engine kind/toil scope/broker version:8.3.0 version:8.3.0-alpha3 +5
   
7. Create a CommandRedistributor zeebe#11914
   component/engine kind/toil version:8.3.0 version:8.3.0-alpha3 +4
   
8. Switch Deployment:Create processor to new record distribution behavior zeebe#11661
   component/engine kind/toil version:8.3.0 version:8.3.0-alpha3 +4
   
9. Implement compact record logger for Command Distribution zeebe#13057
   component/engine kind/toil version:8.3.0 version:8.3.0-alpha3 +4
   
10. Document new Deployment Distribution logic zeebe#13058
    component/engine kind/documentation version:8.3.0 version:8.3.0-alpha3 +4
    
11. Append latest version of records by default zeebe#13161
    component/engine kind/toil version:8.3.0 version:8.3.0-alpha3 +4
    
12. Pause command redistributor when streamprocessor is paused zeebe#13162
    area/ux component/engine component/zeebe kind/toil +4
13. Load balance deployments over partitions zeebe#13163
    area/performance component/engine component/gateway kind/feature +4
14. Segfault on enabling async scheduled task zeebe#13164
    area/reliability component/db component/engine kind/bug severity/high version:8.1.14 version:8.2.8 version:8.3.0 version:8.3.0-alpha3 +9
    
15. Regression in deploying large payloads zeebe#13233
    component/engine good first issue kind/bug onboarding regression scope/broker severity/high version:8.1.14 version:8.2.8 version:8.3.0 version:8.3.0-alpha4 +11
    
16. Document changes to exported records zeebe#13423
    component/exporter kind/documentation target:8.3 +3
    

[korthout]

Moving this to blocked while deciding on the open questions. These are expected to be answered at the start of the upcoming week.

[korthout]

Moved this into in progress. Open questions have either been answered or don't block progress.

[korthout]

Upgrading this issue to EPIC as the breakdown is now clear, and it would be silly to copy the open questions, concept, and tasks breakdown into a new issue just for that.

[remcowesterhoud]

@korthout do we want to close this epic and leave the remaining issue as a standalone improvement?

[korthout]

I'm investigating a small regression (max deployable resource payload reduced from ~2MB to ~1.4MB on multi-partition clusters) due to

• Switch Deployment:Create processor to new record distribution behavior #11661

We can keep it separate but might be good just to track it here until completed.

[korthout]

We can track #13233 separately. Let's close this epic! 🎉