Details
-
Bug
-
Status: Open
-
Major
-
Resolution: Unresolved
-
0.14.0, 0.13.1, 1.0, 1.1, 1.2
-
None
-
None
Description
In the single threaded run loop, a message envelop was replaced using the "tryUpdate(ssp)" function each time before calling ".choose()".
In the AsyncRunLoop that replacement is delayed until the worker is ready and the envelope begins processing. This allows choosing more messages to schedule to tasks which keeps threads busy and allows for scheduling lower priority messages when partitions are available. Good things when threads are available. When a message is chosen for an already processing partition it is added to the task's pendingEnvelopeQueue so that the run loop can choose yet more messages. But the TieredPriorityChooser may respond with a lower priority message for a SystemStreamPartition that is lower than another nonempty SystemStreamPartition since there is no way to exclude a partition or priority level from the ".choose()" operation.
In fact, the Chooser object can be frequently exhausted of all messages. What happens is that for a partition, one messages from every SystemStream that is not empty will be in the FIFO/ArrayDeque pendingEnvelopeQueue which doesn't respect the set priority settings, so we just devolve to a round robin policy.
To reproduce, run a job with the following setting:
task.inputs=kafka.high-priority,kafka.low-priority
job.container.thread.pool.size=2
systems.kafka.streams.high-priority.samza.priority=1
Expected behavior:
Each partition will fully read the high-priority stream before reading messages from the low-priority stream.
Observed behavior:
Each partition reads from both streams as in a round robin policy.