Broken job stream aggregation

[npepinpe]

Describe the bug

From running a benchmark with > 100 workers, it seems we sometimes fail to aggregate streams properly on the gateway and on the broker. In my benchmark, I had 5 job types, and 20 workers (with the exact same configuration) per job type. I would expect to see at most 5 aggregated streams per gateway, but instead some had 30, some had 17, etc.

So there seems to be some race condition or weird thing breaking the aggregation. This has an impact on performance as it means jobs cannot be retried with logically equivalent workers, and must be resubmitted to the engine (possibly).

This also impacts performance because while you think you're scaling out workers, one aggregated stream may not be scaling out, leading to a job being yielded due to being the stream seeming blocked. This means scaling out workers may not yield any performance gain at all, which negates the horizontal scalability goal of Zeebe in that respect.

To Reproduce

Run a benchmark with multiple different job types, with lots of workers per job type. Then, randomly restart your gateways and brokers. After some restarts (sometimes only 1), you will see the aggregation is broken.

Expected behavior

Given 5 job types and 20 workers per type with the exact same config, I would expect to see at most 5 aggregated streams.

Environment:

• Zeebe Version: 8.6.0-SNAPSHOT (and earlier)

[npepinpe]

A quick glance on the gateway shows that the aggregation fails because the streams get a different server stream ID, so it seems the "logical" ID comparison breaks. Unclear yet if it's the metadata or the stream type which causes it to "fail".

It could also be a race condition, but the code path is extremely simple, so I fail to see how that would be possible.

[npepinpe]

Can confirm it's not aggregated properly on either side, broker or gateway. On the gateway clearly the server ID ends up being wrong, but on the broker it's unclear, as the aggregation is done purely over the job type and metadata.

Again, could be there's a race condition on both sides. Having checked the code path, it seems unlikely, as there's a clear single entrypoint wrapped by the actor. Of course, it could be an actor issue (:scream:), but that's unlikely.

[npepinpe]

I tried writing a quick test to reproduce it, but I can't, it always seem to aggregate properly locally.

[npepinpe]

I think the issue is the tenant IDs. I've added them to the endpoint, and you can see the following values:

[
  RemoteJobStream[
    jobType=id-d22ebb23-8ac5-49ee-816a-dc47ca450633, 
    metadata=Metadata[
      worker=command,
      timeout=PT0.5S,
      fetchVariables=[foo, bar],
      tenantIds=[<default>, <default>, <default>, <default>, <default>, <default>, <default>, <default>]
    ], 
    consumers=[
      RemoteStreamId[id=27eb2a1b-d1c2-4033-b804-1cde40101f97, receiver=gateway-1]
    ]
  ],
  RemoteJobStream[
    jobType=id-d22ebb23-8ac5-49ee-816a-dc47ca450633,
    metadata=Metadata[
      worker=command,
      timeout=PT0.5S,
      fetchVariables=[foo, bar],
      tenantIds=[<default>, <default>, <default>, <default>, <default>, <default>, <default>, <default>, <default>]
    ],
    consumers=[
      RemoteStreamId[id=3195f3fd-2be6-4b09-94b3-71a7f7764683, receiver=gateway-1]
    ]
  ]
]

[npepinpe]

The issue is that we reuse the same command builder (StreamJobsCommandImpl), in the JobStreamerImpl, and that one keeps adding the tenants on every send :)