[
https://issues.apache.org/jira/browse/SAMZA-2308?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Dengpan Yin updated SAMZA-2308:
-------------------------------
Affects Version/s: 1.2
1.1
1.0
> AsyncRunLoop and TieredPriorityChooser fail to be used together
> ---------------------------------------------------------------
>
> Key: SAMZA-2308
> URL: https://issues.apache.org/jira/browse/SAMZA-2308
> Project: Samza
> Issue Type: Bug
> Components: container
> Affects Versions: 0.14.0, 0.13.1, 1.0, 1.1, 1.2
> Reporter: Sean McNealy
> Assignee: Dengpan Yin
> Priority: Major
> Attachments: PriorityChooserTest.scala
>
>
> 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.
--
This message was sent by Atlassian Jira
(v8.3.2#803003)
