samza-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Chris Riccomini" <criccom...@apache.org>
Subject Re: Review Request 13725: SAMZA-2
Date Fri, 30 Aug 2013 22:25:35 GMT


> On Aug. 30, 2013, 9:07 p.m., Jay Kreps wrote:
> > This looks good.
> > 
> > What is a situation in which you would ever need a start and stop in the MessageChooser?
Since this is just logic you shouldn't need this, right?
> > 
> > I wonder if this is still a bit high level. Comming to the specific use cases
> > 1. Prioritize by time
> > 2. Bootstrap state first
> > 3. Remain roughly equally behind on each stream
> > 
> > If you think about these. For the first two we haven't really made it easy. For
the third one it still isn't possible.
> > 
> > I kind of feel that making this pluggable is nice, but getting a fully featured
default implementation is more important. This implementation should
> > a. Allow preferring certain streams for the state bootstrap
> > b. Keep all streams of the same priority roughly in sync.
> > 
> > Maybe this is just something each user needs to do once and then reuse that class?

The main motivation for start/stop is allow developers to setup a client that queries some
outside service to make picking decisions. I'm not saying that this is advisable, but I know
people will try and do it. Without stop, there's no way to shut down the client when the service
stops. If we assume the service never stops, then this isn't a problem, but if there is a
definite "end" to the processor (i.e. TaskCoordinator.shutdown), then the chooser needs a
graceful shutdown.

The motivation for register is that there are situations where you want to initialize your
data structures (or whatever) on startup before any messages are received. Letting the MessageChooser
know which SystemStreamPartitions they're going to be receiving messages from just seems like
a good idea. For example, if we wanted to implement Sriram's version of RoundRobinChooser,
you could setup a Map[SystemStreamPartition, ArrayDequeue] on start, instead of having to
lazily check the map and insert a new queue if none exists every time update is called.

(1) can be done by implementing a PriorityChooser that yanks out the timestamp from the message.
I don't think we can do much better than this without adding a "time" field to the IncomingMessageEnvelope,
and making consumers/serdes supply it.
(2) CAN'T be done right now because we don't have any concept of "end of stream" in the IncomingMessageEnvelope.
Since our fetches are asynchronous, the fact that the chooser has no more messages for a given
SSP doesn't actually mean that it's at head. There is no way to tell, and without this, you
can't be guaranteed that you've bootstrapped all state prior to processing messages.
(3) Do you mean equally behind in terms of msgs from head, or equally behind in terms of wall-clock
time? Msgs from head can't be done due to the same reason as (2). Equally behind in wall clock
time is just the same as (1) isn't it?

Correct me if I'm wrong, but it seems like the main issue you have is that we don't have a
way to measure how far behind a message is (messages behind, or milliseconds behind). Without
this, neither (a) or (b) is possible. All we have in the IncomingMessageEnvelope is offset,
but no relation to head (or a timestamp). Without this, I don't think we can satisfy either
criteria, whether we have a good default implementation or not.


- Chris


-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/13725/#review25804
-----------------------------------------------------------


On Aug. 30, 2013, 5:47 p.m., Chris Riccomini wrote:
> 
> -----------------------------------------------------------
> This is an automatically generated e-mail. To reply, visit:
> https://reviews.apache.org/r/13725/
> -----------------------------------------------------------
> 
> (Updated Aug. 30, 2013, 5:47 p.m.)
> 
> 
> Review request for samza.
> 
> 
> Repository: samza
> 
> 
> Description
> -------
> 
> added start, stop, and register to message chooser.
> 
> 
> adding docs for message chooser. swiching round robin chooser back to a queue.
> 
> 
> missed license in message chooser factory
> 
> 
> add apache licensing
> 
> 
> samza container was using message chooser, not message chooser factory. fixed.
> 
> 
> add stream chooser test. update stream chooser to invert priority due to bug.
> 
> 
> add round robin test. fix compile error in round robin chooser.
> 
> 
> add priority chooser test. fix bug in priority chooser that was reversing ordering.
> 
> 
> adding stream chooser. adding message chooser factory.
> 
> 
> adding priority chooser. moving default chooser to round robin chooser. adding config
for chooser
> 
> 
> Diffs
> -----
> 
>   docs/learn/documentation/0.7.0/container/streams.md e755789407b294e02b399e71ba684c1d6dc314c6

>   samza-api/src/main/java/org/apache/samza/system/MessageChooser.java 306b2902303c72f3d7a3eb313f55d7e88d21e00d

>   samza-api/src/main/java/org/apache/samza/system/PriorityChooser.java PRE-CREATION 
>   samza-api/src/test/java/org/apache/samza/system/TestPriorityChooser.java PRE-CREATION

>   samza-core/src/main/scala/org/apache/samza/config/StreamChooserConfig.scala PRE-CREATION

>   samza-core/src/main/scala/org/apache/samza/config/TaskConfig.scala 0c742d83c2f60d2448a79376677713a1ff0b11ec

>   samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala 2d2efdd14c7680c29aad5f2a98349e2fc57cf9fe

>   samza-core/src/main/scala/org/apache/samza/system/DefaultChooser.scala 5a72e7a3bfba0f06a5a98c6ba26865800d7780b9

>   samza-core/src/main/scala/org/apache/samza/system/RoundRobinChooser.scala PRE-CREATION

>   samza-core/src/main/scala/org/apache/samza/system/StreamChooser.scala PRE-CREATION

>   samza-core/src/main/scala/org/apache/samza/system/SystemConsumers.scala b18f0cc5a21088a58db1c26ff43bba06dd3165ac

>   samza-core/src/test/scala/org/apache/samza/system/TestRoundRobinChooser.scala PRE-CREATION

>   samza-core/src/test/scala/org/apache/samza/system/TestStreamChooser.scala PRE-CREATION

>   samza-core/src/test/scala/org/apache/samza/system/TestSystemConsumers.scala PRE-CREATION

> 
> Diff: https://reviews.apache.org/r/13725/diff/
> 
> 
> Testing
> -------
> 
> 
> Thanks,
> 
> Chris Riccomini
> 
>


Mime
  • Unnamed multipart/alternative (inline, None, 0 bytes)
View raw message