You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@samza.apache.org by Chris Riccomini <cr...@apache.org> on 2013/08/28 20:23:19 UTC
Re: Review Request 13725: SAMZA-2
-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/13725/
-----------------------------------------------------------
(Updated Aug. 28, 2013, 6:23 p.m.)
Review request for samza.
Repository: samza
Description (updated)
-------
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 (updated)
-----
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/MessageChooserFactory.java PRE-CREATION
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/StreamChooser.scala PRE-CREATION
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
Diff: https://reviews.apache.org/r/13725/diff/
Testing
-------
Thanks,
Chris Riccomini
Re: Review Request 13725: SAMZA-2
Posted by Chris Riccomini <cr...@apache.org>.
> On Oct. 1, 2013, 4:35 p.m., Sriram Subramanian wrote:
> > One other thing -
> >
> > 1. We have a bunch of features that we want to implement using the chooser.
> >
> > a. bootstrapping
> > b. priority
> > c. batching
> > d. throttle
> >
> > Leaving the features above, the chooser is also used to do what it is meant for, to choose between equal stream partitions based on some heuristic (Example, Round robin).
> >
> > The way the features above should be specified in the config should have no relation to the chooser. For example,
> >
> > task.bootstrap = stream1,stream2
> >
> > task.prioriites.stream1 = 2
> > task.priorities.stream2 = 3
> >
> > task.batch.size = 100
> >
> > task.throttle.stream1 = 10
> >
> > task.message.chooser = RoundRobin
> >
> > All the above features should work irrespective of the chooser specified. Currently, we are using a DefaultChooser that composes all the choosers and orchestrates them. The problem is that the user of the framework will have no idea what the default chooser does. If they replace it with say RoundRobin, all the features above will break. So, instead of using a wrapped.chooser, we should just use the DefaultChooser/ChooserOrchestrator internally but never expose that. The task.message.chooser would determine the default behavior and the rest of the features will just co-exist with whatever chooser that is chosen.
> >
Agree. I was thinking this as well. I'm going to move all of the wiring in DefaultChooserFactory for batch/bootstrap/priority into the SamzaContainer, instead. I'll also move the DefaultChooserConfig stuff into TaskConfig. This will also take care of eliminating task.chooser.wrapped.class (1), as we can just use task.chooser.class instead.
- Chris
-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/13725/#review26564
-----------------------------------------------------------
On Sept. 26, 2013, 3:03 a.m., Chris Riccomini wrote:
>
> -----------------------------------------------------------
> This is an automatically generated e-mail. To reply, visit:
> https://reviews.apache.org/r/13725/
> -----------------------------------------------------------
>
> (Updated Sept. 26, 2013, 3:03 a.m.)
>
>
> Review request for samza.
>
>
> Repository: samza
>
>
> Description
> -------
>
> wrote default chooser test that bootstraps, prioritizes, and batches.
>
>
> turning on priority tests for default chooser. refactor default chooser a bit to make it easier to test.
>
>
> test default chooser with just round robin.
>
>
> bug fix to use round robin chooser factory in default chooser
>
>
> change wiring for default chooser to make it more testable. add start/stop/register tests to all choosers.
>
>
> add one more test in priority chooser.
>
>
> add unit test for tiered priority chooser
>
>
> add unit test for bootstrapping chooser.
>
>
> turn default chooser on in samza container. add license to test stateful task. add more docs to default chooser. clean up batching unit test.
>
>
> add unit test for batching chooser.
>
>
> only wire in wrapping message choosers when we need them. add docs to the default chooser factory.
>
>
> rename class to BootstrappingChooser.
>
>
> refactor to move into org.apache.samza.system.chooser.
>
>
> build latest message offset map.
>
>
> add a streams-behind-chooser that guarantees one message from each SystemStream before choose is called.
>
>
> add start/stop/register back. all tests pass.
>
>
> fix bug -- should allow manual override if Int.MaxInt for bootstrap streams.
>
>
> minor bug in default chooser. was re-using same chooser everywhere.
>
>
> adding wiring in default chooser.
>
>
> initial pass adding composed message choosers.
>
>
> rebase to master, which includes SAMZA-25 metrics. fix several tests that were broken after removing start/stop/register.
>
>
> adding more docs for round robin
>
>
> remove start/stop/register
>
>
> cleanup some wiki markdown in MessageChooser javadoc.
>
>
> add more javadocs to the message chooser.
>
>
> Merge branch 'SAMZA-2_fine-grain-control-over-stream-consumption' of github.com:criccomini/incubator-samza into SAMZA-2_fine-grain-control-over-stream-consumption
>
>
> 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 b15c34d3f68e32d56e0da8af91d78e47a5110f67
> samza-api/src/main/java/org/apache/samza/system/MessageChooser.java 306b2902303c72f3d7a3eb313f55d7e88d21e00d
> samza-api/src/main/java/org/apache/samza/system/SystemAdmin.java c902d414484e05ae75c9ca58ad9629cb01120f62
> samza-api/src/main/java/org/apache/samza/system/chooser/MessageChooser.java PRE-CREATION
> samza-api/src/main/java/org/apache/samza/system/chooser/PriorityChooser.java PRE-CREATION
> samza-api/src/main/java/org/apache/samza/util/SinglePartitionSystemAdmin.java e4ed30bf363142d82a3c40909e160b5825fe60fd
> samza-api/src/test/java/org/apache/samza/system/chooser/TestPriorityChooser.java PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/config/DefaultChooserConfig.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 62bd243115e71612e00784124baa972b33e56cb7
> samza-core/src/main/scala/org/apache/samza/system/DefaultChooser.scala 5a72e7a3bfba0f06a5a98c6ba26865800d7780b9
> samza-core/src/main/scala/org/apache/samza/system/SystemConsumers.scala d24671ec64a42ede6f779effe9c845e1cbbc5e51
> samza-core/src/main/scala/org/apache/samza/system/chooser/BatchingChooser.scala PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/system/chooser/BootstrappingChooser.scala PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/system/chooser/DefaultChooser.scala PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/system/chooser/RoundRobinChooser.scala PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/system/chooser/TieredPriorityChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/TestSystemConsumers.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/MockMessageChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestBatchingChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestBootstrappingChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestDefaultChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestRoundRobinChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestTieredPriorityChooser.scala PRE-CREATION
> samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemAdmin.scala 183c6ccce39dedaef9dba56d5b61ffdedfc9d08a
> samza-test/src/test/scala/org/apache/samza/test/integration/TestStatefulTask.scala 7d4e20a8bdc7a45b0a1b464a6f4b868d1d03eab0
> samza-yarn/src/test/scala/org/apache/samza/job/yarn/TestSamzaAppMasterTaskManager.scala 68050528cd5f8acfe3a1f7563b4e7fe6c7473be5
>
> Diff: https://reviews.apache.org/r/13725/diff/
>
>
> Testing
> -------
>
>
> Thanks,
>
> Chris Riccomini
>
>
Re: Review Request 13725: SAMZA-2
Posted by Sriram Subramanian <sr...@gmail.com>.
-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/13725/#review26564
-----------------------------------------------------------
One other thing -
1. We have a bunch of features that we want to implement using the chooser.
a. bootstrapping
b. priority
c. batching
d. throttle
Leaving the features above, the chooser is also used to do what it is meant for, to choose between equal stream partitions based on some heuristic (Example, Round robin).
The way the features above should be specified in the config should have no relation to the chooser. For example,
task.bootstrap = stream1,stream2
task.prioriites.stream1 = 2
task.priorities.stream2 = 3
task.batch.size = 100
task.throttle.stream1 = 10
task.message.chooser = RoundRobin
All the above features should work irrespective of the chooser specified. Currently, we are using a DefaultChooser that composes all the choosers and orchestrates them. The problem is that the user of the framework will have no idea what the default chooser does. If they replace it with say RoundRobin, all the features above will break. So, instead of using a wrapped.chooser, we should just use the DefaultChooser/ChooserOrchestrator internally but never expose that. The task.message.chooser would determine the default behavior and the rest of the features will just co-exist with whatever chooser that is chosen.
- Sriram Subramanian
On Sept. 26, 2013, 3:03 a.m., Chris Riccomini wrote:
>
> -----------------------------------------------------------
> This is an automatically generated e-mail. To reply, visit:
> https://reviews.apache.org/r/13725/
> -----------------------------------------------------------
>
> (Updated Sept. 26, 2013, 3:03 a.m.)
>
>
> Review request for samza.
>
>
> Repository: samza
>
>
> Description
> -------
>
> wrote default chooser test that bootstraps, prioritizes, and batches.
>
>
> turning on priority tests for default chooser. refactor default chooser a bit to make it easier to test.
>
>
> test default chooser with just round robin.
>
>
> bug fix to use round robin chooser factory in default chooser
>
>
> change wiring for default chooser to make it more testable. add start/stop/register tests to all choosers.
>
>
> add one more test in priority chooser.
>
>
> add unit test for tiered priority chooser
>
>
> add unit test for bootstrapping chooser.
>
>
> turn default chooser on in samza container. add license to test stateful task. add more docs to default chooser. clean up batching unit test.
>
>
> add unit test for batching chooser.
>
>
> only wire in wrapping message choosers when we need them. add docs to the default chooser factory.
>
>
> rename class to BootstrappingChooser.
>
>
> refactor to move into org.apache.samza.system.chooser.
>
>
> build latest message offset map.
>
>
> add a streams-behind-chooser that guarantees one message from each SystemStream before choose is called.
>
>
> add start/stop/register back. all tests pass.
>
>
> fix bug -- should allow manual override if Int.MaxInt for bootstrap streams.
>
>
> minor bug in default chooser. was re-using same chooser everywhere.
>
>
> adding wiring in default chooser.
>
>
> initial pass adding composed message choosers.
>
>
> rebase to master, which includes SAMZA-25 metrics. fix several tests that were broken after removing start/stop/register.
>
>
> adding more docs for round robin
>
>
> remove start/stop/register
>
>
> cleanup some wiki markdown in MessageChooser javadoc.
>
>
> add more javadocs to the message chooser.
>
>
> Merge branch 'SAMZA-2_fine-grain-control-over-stream-consumption' of github.com:criccomini/incubator-samza into SAMZA-2_fine-grain-control-over-stream-consumption
>
>
> 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 b15c34d3f68e32d56e0da8af91d78e47a5110f67
> samza-api/src/main/java/org/apache/samza/system/MessageChooser.java 306b2902303c72f3d7a3eb313f55d7e88d21e00d
> samza-api/src/main/java/org/apache/samza/system/SystemAdmin.java c902d414484e05ae75c9ca58ad9629cb01120f62
> samza-api/src/main/java/org/apache/samza/system/chooser/MessageChooser.java PRE-CREATION
> samza-api/src/main/java/org/apache/samza/system/chooser/PriorityChooser.java PRE-CREATION
> samza-api/src/main/java/org/apache/samza/util/SinglePartitionSystemAdmin.java e4ed30bf363142d82a3c40909e160b5825fe60fd
> samza-api/src/test/java/org/apache/samza/system/chooser/TestPriorityChooser.java PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/config/DefaultChooserConfig.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 62bd243115e71612e00784124baa972b33e56cb7
> samza-core/src/main/scala/org/apache/samza/system/DefaultChooser.scala 5a72e7a3bfba0f06a5a98c6ba26865800d7780b9
> samza-core/src/main/scala/org/apache/samza/system/SystemConsumers.scala d24671ec64a42ede6f779effe9c845e1cbbc5e51
> samza-core/src/main/scala/org/apache/samza/system/chooser/BatchingChooser.scala PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/system/chooser/BootstrappingChooser.scala PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/system/chooser/DefaultChooser.scala PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/system/chooser/RoundRobinChooser.scala PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/system/chooser/TieredPriorityChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/TestSystemConsumers.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/MockMessageChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestBatchingChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestBootstrappingChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestDefaultChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestRoundRobinChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestTieredPriorityChooser.scala PRE-CREATION
> samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemAdmin.scala 183c6ccce39dedaef9dba56d5b61ffdedfc9d08a
> samza-test/src/test/scala/org/apache/samza/test/integration/TestStatefulTask.scala 7d4e20a8bdc7a45b0a1b464a6f4b868d1d03eab0
> samza-yarn/src/test/scala/org/apache/samza/job/yarn/TestSamzaAppMasterTaskManager.scala 68050528cd5f8acfe3a1f7563b4e7fe6c7473be5
>
> Diff: https://reviews.apache.org/r/13725/diff/
>
>
> Testing
> -------
>
>
> Thanks,
>
> Chris Riccomini
>
>
Re: Review Request 13725: SAMZA-2
Posted by Chris Riccomini <cr...@apache.org>.
> On Oct. 2, 2013, 6:01 p.m., Sriram Subramanian wrote:
> > streams.md
> >
> > 1.1 Need to remove this line. "Samza also provides a helper class called PriorityChooser which lets you choose which message to process next based on a priority score."
> >
> > MessageChooser.java
> >
> > 1.1 comment fix - "lived systems"?
> > 1.2 comment fix - "checkpointd"
> >
> > WrappedChooserConfig.java
> >
> > 1.1 I think these names should not have anything to do with the chooser.
> > "task.chooser.bootstrap.%s.%s", "task.chooser.priorities.%s.%s", "task.chooser.batch.size"
> >
> > 1.2 I really think Wrapped is not a great name here. The name should give an idea what the chooser does.
> >
> > BatchingChooser.java
> >
> > 1.1 comment fix - "patching"
> > 1.2 if you decide to keep the main choosers name as Wrapped, the context of wrapped chooser in the comments is confusing
> >
> > BootstrappingChooser.java
> >
> > 1.1 comment fix - "at least on envelope for every"
> > 1.2 Why do we need to wait for every stream in latestMessageOffsets to get atleast one event? We just need to wait till atleast one stream in latestMessageOffsets get an event. No?
> > 1.3 Same comment here about the naming of wrapped. Confusing name with the main class that orchestrates things.
> >
> > WrappedChooser.java
> >
> > 1.1 We currently override the priority of bootstrap streams with the explicit priority. This is actually confusing. I think we should just fail with config error when user overrides the bootstrap streams priority. We should not start anything without bootstrapping all the streams.
> >
> > KafkaSystemAdminCommand.java
> >
> > 1.1 Why are you not using getTopicInfo to get the metadata info for the topics?
Regarding WrappedChooser.java:1.1, and BootstrappingChooser.java:1.2, they're actually related. By default, all bootstrap streams are prioritized the same. When that's the case, then we don't need to wait for all SystemStreams to have one message (1.2), since it's just as valid to process any bootstrap stream.
If, however, a non-random strategy is required when choosing between bootstrap streams, then you need to provide a message from every SystemStream in order for the wrapped chooser to make the correct decision. Two examples of a non-random choosing strategy would be to prioritize streams based on time (keeping the bootstrap streams time aligned), or hard-coding a priority for the bootstrap streams (e.g. bootstrap stream A = 2, bootstrap stream B = 1).
So, if we take (1.2) as a requirement to provide generic bootstrapping with non-random strategies, the next question is, does it make sense to allow non-random picking between bootstrap streams? I think it does. One example of this is bootstrap streams with differing priorities (mentioned above). By prioritizing bootstrap stream A over bootstrap stream B, you bootstrap all of stream A before bootstrapping any of stream B. This behavior might be desirable in cases where you're doing a join from stream B to stream A and creating a materialized view of the join. Rather than bootstrapping stream A and stream B simultaneously, and then doing the join after both bootstraps are complete, you could bootstrap only stream A, then as messages from stream B come in, join to the stream A data, and produce the materialized view on the fly. For example, you might have a profiles bootstrap stream and a companies bootstrap stream, and you might want to produce a table with a user's name and the compa
ny they work for. If you set profiles to priority 2, and companies to priority 1, you can load profiles into a KV store, then take each company event, join it with the profiles KV store (to get the user's name), and then write directly to the "name and company" KV store. If you don't allow prioritized bootstrap streams, then you might get company events before the profile, and you have to use two KV stores to do the join. It can still be done on the fly, but it's more cumbersome.
The time-aligned case is similar. If you want to keep your bootstrap streams (roughly) time-aligned with each other, you can do a much better job of this if you look at the next message from each bootstrap stream, rather than taking the first one that comes in.
The trade-off of this style is that it's going to introduce more latency during bootstrap in the common case, since it's valid to process any arbitrary bootstrap stream envelopes if they're all the same priority.
It's unclear to me whether these two examples are going to be edge cases, or more common styles of processing. It's also unclear if there are more use cases for non-random bootstrap prioritization. Given these unknowns, I'd rather be more general, and impact latency a bit, rather than improve latency, and disabling prioritized bootstrap streams out-right.
- Chris
-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/13725/#review26614
-----------------------------------------------------------
On Oct. 2, 2013, 12:49 a.m., Chris Riccomini wrote:
>
> -----------------------------------------------------------
> This is an automatically generated e-mail. To reply, visit:
> https://reviews.apache.org/r/13725/
> -----------------------------------------------------------
>
> (Updated Oct. 2, 2013, 12:49 a.m.)
>
>
> Review request for samza.
>
>
> Repository: samza
>
>
> Description
> -------
>
> adding a first pass at KafkaSystemAdmin.getLastOffsets
>
>
> use the systemAdmins from samza container in wrapped chooser, rather than doing wiring internally.
>
>
> changing ordering of wrapped chooser from bootstrap -> priority -> batching to bootstrap -> batching -> priority.
>
>
> switch default chooser to WrappedChooser. Always use wrapped chooser. Remove task.chooser.wrapped.class.
>
>
> wrote default chooser test that bootstraps, prioritizes, and batches.
>
>
> turning on priority tests for default chooser. refactor default chooser a bit to make it easier to test.
>
>
> test default chooser with just round robin.
>
>
> bug fix to use round robin chooser factory in default chooser
>
>
> change wiring for default chooser to make it more testable. add start/stop/register tests to all choosers.
>
>
> add one more test in priority chooser.
>
>
> add unit test for tiered priority chooser
>
>
> add unit test for bootstrapping chooser.
>
>
> turn default chooser on in samza container. add license to test stateful task. add more docs to default chooser. clean up batching unit test.
>
>
> add unit test for batching chooser.
>
>
> only wire in wrapping message choosers when we need them. add docs to the default chooser factory.
>
>
> rename class to BootstrappingChooser.
>
>
> refactor to move into org.apache.samza.system.chooser.
>
>
> build latest message offset map.
>
>
> add a streams-behind-chooser that guarantees one message from each SystemStream before choose is called.
>
>
> add start/stop/register back. all tests pass.
>
>
> fix bug -- should allow manual override if Int.MaxInt for bootstrap streams.
>
>
> minor bug in default chooser. was re-using same chooser everywhere.
>
>
> adding wiring in default chooser.
>
>
> initial pass adding composed message choosers.
>
>
> rebase to master, which includes SAMZA-25 metrics. fix several tests that were broken after removing start/stop/register.
>
>
> adding more docs for round robin
>
>
> remove start/stop/register
>
>
> cleanup some wiki markdown in MessageChooser javadoc.
>
>
> add more javadocs to the message chooser.
>
>
> Merge branch 'SAMZA-2_fine-grain-control-over-stream-consumption' of github.com:criccomini/incubator-samza into SAMZA-2_fine-grain-control-over-stream-consumption
>
>
> 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 b15c34d3f68e32d56e0da8af91d78e47a5110f67
> samza-api/src/main/java/org/apache/samza/system/MessageChooser.java 306b2902303c72f3d7a3eb313f55d7e88d21e00d
> samza-api/src/main/java/org/apache/samza/system/SystemAdmin.java c902d414484e05ae75c9ca58ad9629cb01120f62
> samza-api/src/main/java/org/apache/samza/system/chooser/MessageChooser.java PRE-CREATION
> samza-api/src/main/java/org/apache/samza/util/SinglePartitionSystemAdmin.java e4ed30bf363142d82a3c40909e160b5825fe60fd
> samza-core/src/main/scala/org/apache/samza/config/TaskConfig.scala 0c742d83c2f60d2448a79376677713a1ff0b11ec
> samza-core/src/main/scala/org/apache/samza/config/WrappedChooserConfig.scala PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala 62bd243115e71612e00784124baa972b33e56cb7
> samza-core/src/main/scala/org/apache/samza/system/DefaultChooser.scala 5a72e7a3bfba0f06a5a98c6ba26865800d7780b9
> samza-core/src/main/scala/org/apache/samza/system/SystemConsumers.scala d24671ec64a42ede6f779effe9c845e1cbbc5e51
> samza-core/src/main/scala/org/apache/samza/system/chooser/BatchingChooser.scala PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/system/chooser/BootstrappingChooser.scala PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/system/chooser/RoundRobinChooser.scala PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/system/chooser/TieredPriorityChooser.scala PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/system/chooser/WrappedChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/TestSystemConsumers.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/MockMessageChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestBatchingChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestBootstrappingChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestRoundRobinChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestTieredPriorityChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestWrappedChooser.scala PRE-CREATION
> samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemAdmin.scala 183c6ccce39dedaef9dba56d5b61ffdedfc9d08a
> samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemFactory.scala ba08af85892f0793d01e3118fa7ad6d569a0f238
> samza-test/src/test/scala/org/apache/samza/test/integration/TestStatefulTask.scala 7d4e20a8bdc7a45b0a1b464a6f4b868d1d03eab0
> samza-yarn/src/test/scala/org/apache/samza/job/yarn/TestSamzaAppMasterTaskManager.scala 68050528cd5f8acfe3a1f7563b4e7fe6c7473be5
>
> Diff: https://reviews.apache.org/r/13725/diff/
>
>
> Testing
> -------
>
>
> Thanks,
>
> Chris Riccomini
>
>
Re: Review Request 13725: SAMZA-2
Posted by Sriram Subramanian <sr...@gmail.com>.
-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/13725/#review26614
-----------------------------------------------------------
streams.md
1.1 Need to remove this line. "Samza also provides a helper class called PriorityChooser which lets you choose which message to process next based on a priority score."
MessageChooser.java
1.1 comment fix - "lived systems"?
1.2 comment fix - "checkpointd"
WrappedChooserConfig.java
1.1 I think these names should not have anything to do with the chooser.
"task.chooser.bootstrap.%s.%s", "task.chooser.priorities.%s.%s", "task.chooser.batch.size"
1.2 I really think Wrapped is not a great name here. The name should give an idea what the chooser does.
BatchingChooser.java
1.1 comment fix - "patching"
1.2 if you decide to keep the main choosers name as Wrapped, the context of wrapped chooser in the comments is confusing
BootstrappingChooser.java
1.1 comment fix - "at least on envelope for every"
1.2 Why do we need to wait for every stream in latestMessageOffsets to get atleast one event? We just need to wait till atleast one stream in latestMessageOffsets get an event. No?
1.3 Same comment here about the naming of wrapped. Confusing name with the main class that orchestrates things.
WrappedChooser.java
1.1 We currently override the priority of bootstrap streams with the explicit priority. This is actually confusing. I think we should just fail with config error when user overrides the bootstrap streams priority. We should not start anything without bootstrapping all the streams.
KafkaSystemAdminCommand.java
1.1 Why are you not using getTopicInfo to get the metadata info for the topics?
- Sriram Subramanian
On Oct. 2, 2013, 12:49 a.m., Chris Riccomini wrote:
>
> -----------------------------------------------------------
> This is an automatically generated e-mail. To reply, visit:
> https://reviews.apache.org/r/13725/
> -----------------------------------------------------------
>
> (Updated Oct. 2, 2013, 12:49 a.m.)
>
>
> Review request for samza.
>
>
> Repository: samza
>
>
> Description
> -------
>
> adding a first pass at KafkaSystemAdmin.getLastOffsets
>
>
> use the systemAdmins from samza container in wrapped chooser, rather than doing wiring internally.
>
>
> changing ordering of wrapped chooser from bootstrap -> priority -> batching to bootstrap -> batching -> priority.
>
>
> switch default chooser to WrappedChooser. Always use wrapped chooser. Remove task.chooser.wrapped.class.
>
>
> wrote default chooser test that bootstraps, prioritizes, and batches.
>
>
> turning on priority tests for default chooser. refactor default chooser a bit to make it easier to test.
>
>
> test default chooser with just round robin.
>
>
> bug fix to use round robin chooser factory in default chooser
>
>
> change wiring for default chooser to make it more testable. add start/stop/register tests to all choosers.
>
>
> add one more test in priority chooser.
>
>
> add unit test for tiered priority chooser
>
>
> add unit test for bootstrapping chooser.
>
>
> turn default chooser on in samza container. add license to test stateful task. add more docs to default chooser. clean up batching unit test.
>
>
> add unit test for batching chooser.
>
>
> only wire in wrapping message choosers when we need them. add docs to the default chooser factory.
>
>
> rename class to BootstrappingChooser.
>
>
> refactor to move into org.apache.samza.system.chooser.
>
>
> build latest message offset map.
>
>
> add a streams-behind-chooser that guarantees one message from each SystemStream before choose is called.
>
>
> add start/stop/register back. all tests pass.
>
>
> fix bug -- should allow manual override if Int.MaxInt for bootstrap streams.
>
>
> minor bug in default chooser. was re-using same chooser everywhere.
>
>
> adding wiring in default chooser.
>
>
> initial pass adding composed message choosers.
>
>
> rebase to master, which includes SAMZA-25 metrics. fix several tests that were broken after removing start/stop/register.
>
>
> adding more docs for round robin
>
>
> remove start/stop/register
>
>
> cleanup some wiki markdown in MessageChooser javadoc.
>
>
> add more javadocs to the message chooser.
>
>
> Merge branch 'SAMZA-2_fine-grain-control-over-stream-consumption' of github.com:criccomini/incubator-samza into SAMZA-2_fine-grain-control-over-stream-consumption
>
>
> 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 b15c34d3f68e32d56e0da8af91d78e47a5110f67
> samza-api/src/main/java/org/apache/samza/system/MessageChooser.java 306b2902303c72f3d7a3eb313f55d7e88d21e00d
> samza-api/src/main/java/org/apache/samza/system/SystemAdmin.java c902d414484e05ae75c9ca58ad9629cb01120f62
> samza-api/src/main/java/org/apache/samza/system/chooser/MessageChooser.java PRE-CREATION
> samza-api/src/main/java/org/apache/samza/util/SinglePartitionSystemAdmin.java e4ed30bf363142d82a3c40909e160b5825fe60fd
> samza-core/src/main/scala/org/apache/samza/config/TaskConfig.scala 0c742d83c2f60d2448a79376677713a1ff0b11ec
> samza-core/src/main/scala/org/apache/samza/config/WrappedChooserConfig.scala PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala 62bd243115e71612e00784124baa972b33e56cb7
> samza-core/src/main/scala/org/apache/samza/system/DefaultChooser.scala 5a72e7a3bfba0f06a5a98c6ba26865800d7780b9
> samza-core/src/main/scala/org/apache/samza/system/SystemConsumers.scala d24671ec64a42ede6f779effe9c845e1cbbc5e51
> samza-core/src/main/scala/org/apache/samza/system/chooser/BatchingChooser.scala PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/system/chooser/BootstrappingChooser.scala PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/system/chooser/RoundRobinChooser.scala PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/system/chooser/TieredPriorityChooser.scala PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/system/chooser/WrappedChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/TestSystemConsumers.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/MockMessageChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestBatchingChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestBootstrappingChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestRoundRobinChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestTieredPriorityChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestWrappedChooser.scala PRE-CREATION
> samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemAdmin.scala 183c6ccce39dedaef9dba56d5b61ffdedfc9d08a
> samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemFactory.scala ba08af85892f0793d01e3118fa7ad6d569a0f238
> samza-test/src/test/scala/org/apache/samza/test/integration/TestStatefulTask.scala 7d4e20a8bdc7a45b0a1b464a6f4b868d1d03eab0
> samza-yarn/src/test/scala/org/apache/samza/job/yarn/TestSamzaAppMasterTaskManager.scala 68050528cd5f8acfe3a1f7563b4e7fe6c7473be5
>
> Diff: https://reviews.apache.org/r/13725/diff/
>
>
> Testing
> -------
>
>
> Thanks,
>
> Chris Riccomini
>
>
Re: Review Request 13725: SAMZA-2
Posted by Sriram Subramanian <sr...@gmail.com>.
-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/13725/#review26699
-----------------------------------------------------------
Ship it!
Ship It!
- Sriram Subramanian
On Oct. 3, 2013, 9:32 p.m., Chris Riccomini wrote:
>
> -----------------------------------------------------------
> This is an automatically generated e-mail. To reply, visit:
> https://reviews.apache.org/r/13725/
> -----------------------------------------------------------
>
> (Updated Oct. 3, 2013, 9:32 p.m.)
>
>
> Review request for samza.
>
>
> Repository: samza
>
>
> Description
> -------
>
> adding docs for default chooser.
>
>
> adding metrics and logging.
>
>
> Adding metrics to round robin chooser. updating MessageChooserFactory to take a metrics registry.
>
>
> adding logs to kafka system admin
>
>
> handle interrups in KafkaSystemAdmin.
>
>
> rename wrappedchooser back to default chooser. fix some comments.
>
>
> use topic metadata cache to get metadata
>
>
> adding test for kafkasystemadmin.getlatestoffsets. bumping up gradle heap for samsa-test and samsa-kafka tests. shutting down zookeeper in tests, so we don't leak ports.
>
>
> adding a first pass at KafkaSystemAdmin.getLastOffsets
>
>
> use the systemAdmins from samza container in wrapped chooser, rather than doing wiring internally.
>
>
> changing ordering of wrapped chooser from bootstrap -> priority -> batching to bootstrap -> batching -> priority.
>
>
> switch default chooser to WrappedChooser. Always use wrapped chooser. Remove task.chooser.wrapped.class.
>
>
> wrote default chooser test that bootstraps, prioritizes, and batches.
>
>
> turning on priority tests for default chooser. refactor default chooser a bit to make it easier to test.
>
>
> test default chooser with just round robin.
>
>
> bug fix to use round robin chooser factory in default chooser
>
>
> change wiring for default chooser to make it more testable. add start/stop/register tests to all choosers.
>
>
> add one more test in priority chooser.
>
>
> add unit test for tiered priority chooser
>
>
> add unit test for bootstrapping chooser.
>
>
> turn default chooser on in samza container. add license to test stateful task. add more docs to default chooser. clean up batching unit test.
>
>
> add unit test for batching chooser.
>
>
> only wire in wrapping message choosers when we need them. add docs to the default chooser factory.
>
>
> rename class to BootstrappingChooser.
>
>
> refactor to move into org.apache.samza.system.chooser.
>
>
> build latest message offset map.
>
>
> add a streams-behind-chooser that guarantees one message from each SystemStream before choose is called.
>
>
> add start/stop/register back. all tests pass.
>
>
> fix bug -- should allow manual override if Int.MaxInt for bootstrap streams.
>
>
> minor bug in default chooser. was re-using same chooser everywhere.
>
>
> adding wiring in default chooser.
>
>
> initial pass adding composed message choosers.
>
>
> rebase to master, which includes SAMZA-25 metrics. fix several tests that were broken after removing start/stop/register.
>
>
> adding more docs for round robin
>
>
> remove start/stop/register
>
>
> cleanup some wiki markdown in MessageChooser javadoc.
>
>
> add more javadocs to the message chooser.
>
>
> Merge branch 'SAMZA-2_fine-grain-control-over-stream-consumption' of github.com:criccomini/incubator-samza into SAMZA-2_fine-grain-control-over-stream-consumption
>
>
> 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
> -----
>
> build.gradle 077e44b0ed521504096b47f41b6180449e97280e
> docs/learn/documentation/0.7.0/container/streams.md b15c34d3f68e32d56e0da8af91d78e47a5110f67
> samza-api/src/main/java/org/apache/samza/system/MessageChooser.java 306b2902303c72f3d7a3eb313f55d7e88d21e00d
> samza-api/src/main/java/org/apache/samza/system/SystemAdmin.java c902d414484e05ae75c9ca58ad9629cb01120f62
> samza-api/src/main/java/org/apache/samza/system/chooser/MessageChooser.java PRE-CREATION
> samza-api/src/main/java/org/apache/samza/util/SinglePartitionSystemAdmin.java e4ed30bf363142d82a3c40909e160b5825fe60fd
> samza-core/src/main/scala/org/apache/samza/config/DefaultChooserConfig.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 62bd243115e71612e00784124baa972b33e56cb7
> samza-core/src/main/scala/org/apache/samza/system/DefaultChooser.scala 5a72e7a3bfba0f06a5a98c6ba26865800d7780b9
> samza-core/src/main/scala/org/apache/samza/system/SystemConsumers.scala d24671ec64a42ede6f779effe9c845e1cbbc5e51
> samza-core/src/main/scala/org/apache/samza/system/chooser/BatchingChooser.scala PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/system/chooser/BootstrappingChooser.scala PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/system/chooser/DefaultChooser.scala PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/system/chooser/RoundRobinChooser.scala PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/system/chooser/TieredPriorityChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/TestSystemConsumers.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/MockMessageChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestBatchingChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestBootstrappingChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestDefaultChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestRoundRobinChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestTieredPriorityChooser.scala PRE-CREATION
> samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemAdmin.scala 183c6ccce39dedaef9dba56d5b61ffdedfc9d08a
> samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemFactory.scala ba08af85892f0793d01e3118fa7ad6d569a0f238
> samza-kafka/src/test/scala/org/apache/samza/checkpoint/TestKafkaCheckpointManager.scala baf4695108b62fb2826cf0d9cdc484ba6d7bb524
> samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestKafkaSystemAdmin.scala PRE-CREATION
> samza-test/src/test/scala/org/apache/samza/test/integration/TestStatefulTask.scala 7d4e20a8bdc7a45b0a1b464a6f4b868d1d03eab0
> samza-yarn/src/test/scala/org/apache/samza/job/yarn/TestSamzaAppMasterTaskManager.scala 68050528cd5f8acfe3a1f7563b4e7fe6c7473be5
>
> Diff: https://reviews.apache.org/r/13725/diff/
>
>
> Testing
> -------
>
>
> Thanks,
>
> Chris Riccomini
>
>
Re: Review Request 13725: SAMZA-2
Posted by Chris Riccomini <cr...@apache.org>.
-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/13725/
-----------------------------------------------------------
(Updated Oct. 3, 2013, 9:32 p.m.)
Review request for samza.
Repository: samza
Description (updated)
-------
adding docs for default chooser.
adding metrics and logging.
Adding metrics to round robin chooser. updating MessageChooserFactory to take a metrics registry.
adding logs to kafka system admin
handle interrups in KafkaSystemAdmin.
rename wrappedchooser back to default chooser. fix some comments.
use topic metadata cache to get metadata
adding test for kafkasystemadmin.getlatestoffsets. bumping up gradle heap for samsa-test and samsa-kafka tests. shutting down zookeeper in tests, so we don't leak ports.
adding a first pass at KafkaSystemAdmin.getLastOffsets
use the systemAdmins from samza container in wrapped chooser, rather than doing wiring internally.
changing ordering of wrapped chooser from bootstrap -> priority -> batching to bootstrap -> batching -> priority.
switch default chooser to WrappedChooser. Always use wrapped chooser. Remove task.chooser.wrapped.class.
wrote default chooser test that bootstraps, prioritizes, and batches.
turning on priority tests for default chooser. refactor default chooser a bit to make it easier to test.
test default chooser with just round robin.
bug fix to use round robin chooser factory in default chooser
change wiring for default chooser to make it more testable. add start/stop/register tests to all choosers.
add one more test in priority chooser.
add unit test for tiered priority chooser
add unit test for bootstrapping chooser.
turn default chooser on in samza container. add license to test stateful task. add more docs to default chooser. clean up batching unit test.
add unit test for batching chooser.
only wire in wrapping message choosers when we need them. add docs to the default chooser factory.
rename class to BootstrappingChooser.
refactor to move into org.apache.samza.system.chooser.
build latest message offset map.
add a streams-behind-chooser that guarantees one message from each SystemStream before choose is called.
add start/stop/register back. all tests pass.
fix bug -- should allow manual override if Int.MaxInt for bootstrap streams.
minor bug in default chooser. was re-using same chooser everywhere.
adding wiring in default chooser.
initial pass adding composed message choosers.
rebase to master, which includes SAMZA-25 metrics. fix several tests that were broken after removing start/stop/register.
adding more docs for round robin
remove start/stop/register
cleanup some wiki markdown in MessageChooser javadoc.
add more javadocs to the message chooser.
Merge branch 'SAMZA-2_fine-grain-control-over-stream-consumption' of github.com:criccomini/incubator-samza into SAMZA-2_fine-grain-control-over-stream-consumption
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 (updated)
-----
build.gradle 077e44b0ed521504096b47f41b6180449e97280e
docs/learn/documentation/0.7.0/container/streams.md b15c34d3f68e32d56e0da8af91d78e47a5110f67
samza-api/src/main/java/org/apache/samza/system/MessageChooser.java 306b2902303c72f3d7a3eb313f55d7e88d21e00d
samza-api/src/main/java/org/apache/samza/system/SystemAdmin.java c902d414484e05ae75c9ca58ad9629cb01120f62
samza-api/src/main/java/org/apache/samza/system/chooser/MessageChooser.java PRE-CREATION
samza-api/src/main/java/org/apache/samza/util/SinglePartitionSystemAdmin.java e4ed30bf363142d82a3c40909e160b5825fe60fd
samza-core/src/main/scala/org/apache/samza/config/DefaultChooserConfig.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 62bd243115e71612e00784124baa972b33e56cb7
samza-core/src/main/scala/org/apache/samza/system/DefaultChooser.scala 5a72e7a3bfba0f06a5a98c6ba26865800d7780b9
samza-core/src/main/scala/org/apache/samza/system/SystemConsumers.scala d24671ec64a42ede6f779effe9c845e1cbbc5e51
samza-core/src/main/scala/org/apache/samza/system/chooser/BatchingChooser.scala PRE-CREATION
samza-core/src/main/scala/org/apache/samza/system/chooser/BootstrappingChooser.scala PRE-CREATION
samza-core/src/main/scala/org/apache/samza/system/chooser/DefaultChooser.scala PRE-CREATION
samza-core/src/main/scala/org/apache/samza/system/chooser/RoundRobinChooser.scala PRE-CREATION
samza-core/src/main/scala/org/apache/samza/system/chooser/TieredPriorityChooser.scala PRE-CREATION
samza-core/src/test/scala/org/apache/samza/system/TestSystemConsumers.scala PRE-CREATION
samza-core/src/test/scala/org/apache/samza/system/chooser/MockMessageChooser.scala PRE-CREATION
samza-core/src/test/scala/org/apache/samza/system/chooser/TestBatchingChooser.scala PRE-CREATION
samza-core/src/test/scala/org/apache/samza/system/chooser/TestBootstrappingChooser.scala PRE-CREATION
samza-core/src/test/scala/org/apache/samza/system/chooser/TestDefaultChooser.scala PRE-CREATION
samza-core/src/test/scala/org/apache/samza/system/chooser/TestRoundRobinChooser.scala PRE-CREATION
samza-core/src/test/scala/org/apache/samza/system/chooser/TestTieredPriorityChooser.scala PRE-CREATION
samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemAdmin.scala 183c6ccce39dedaef9dba56d5b61ffdedfc9d08a
samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemFactory.scala ba08af85892f0793d01e3118fa7ad6d569a0f238
samza-kafka/src/test/scala/org/apache/samza/checkpoint/TestKafkaCheckpointManager.scala baf4695108b62fb2826cf0d9cdc484ba6d7bb524
samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestKafkaSystemAdmin.scala PRE-CREATION
samza-test/src/test/scala/org/apache/samza/test/integration/TestStatefulTask.scala 7d4e20a8bdc7a45b0a1b464a6f4b868d1d03eab0
samza-yarn/src/test/scala/org/apache/samza/job/yarn/TestSamzaAppMasterTaskManager.scala 68050528cd5f8acfe3a1f7563b4e7fe6c7473be5
Diff: https://reviews.apache.org/r/13725/diff/
Testing
-------
Thanks,
Chris Riccomini
Re: Review Request 13725: SAMZA-2
Posted by Chris Riccomini <cr...@apache.org>.
-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/13725/
-----------------------------------------------------------
(Updated Oct. 2, 2013, 12:49 a.m.)
Review request for samza.
Repository: samza
Description (updated)
-------
adding a first pass at KafkaSystemAdmin.getLastOffsets
use the systemAdmins from samza container in wrapped chooser, rather than doing wiring internally.
changing ordering of wrapped chooser from bootstrap -> priority -> batching to bootstrap -> batching -> priority.
switch default chooser to WrappedChooser. Always use wrapped chooser. Remove task.chooser.wrapped.class.
wrote default chooser test that bootstraps, prioritizes, and batches.
turning on priority tests for default chooser. refactor default chooser a bit to make it easier to test.
test default chooser with just round robin.
bug fix to use round robin chooser factory in default chooser
change wiring for default chooser to make it more testable. add start/stop/register tests to all choosers.
add one more test in priority chooser.
add unit test for tiered priority chooser
add unit test for bootstrapping chooser.
turn default chooser on in samza container. add license to test stateful task. add more docs to default chooser. clean up batching unit test.
add unit test for batching chooser.
only wire in wrapping message choosers when we need them. add docs to the default chooser factory.
rename class to BootstrappingChooser.
refactor to move into org.apache.samza.system.chooser.
build latest message offset map.
add a streams-behind-chooser that guarantees one message from each SystemStream before choose is called.
add start/stop/register back. all tests pass.
fix bug -- should allow manual override if Int.MaxInt for bootstrap streams.
minor bug in default chooser. was re-using same chooser everywhere.
adding wiring in default chooser.
initial pass adding composed message choosers.
rebase to master, which includes SAMZA-25 metrics. fix several tests that were broken after removing start/stop/register.
adding more docs for round robin
remove start/stop/register
cleanup some wiki markdown in MessageChooser javadoc.
add more javadocs to the message chooser.
Merge branch 'SAMZA-2_fine-grain-control-over-stream-consumption' of github.com:criccomini/incubator-samza into SAMZA-2_fine-grain-control-over-stream-consumption
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 (updated)
-----
docs/learn/documentation/0.7.0/container/streams.md b15c34d3f68e32d56e0da8af91d78e47a5110f67
samza-api/src/main/java/org/apache/samza/system/MessageChooser.java 306b2902303c72f3d7a3eb313f55d7e88d21e00d
samza-api/src/main/java/org/apache/samza/system/SystemAdmin.java c902d414484e05ae75c9ca58ad9629cb01120f62
samza-api/src/main/java/org/apache/samza/system/chooser/MessageChooser.java PRE-CREATION
samza-api/src/main/java/org/apache/samza/util/SinglePartitionSystemAdmin.java e4ed30bf363142d82a3c40909e160b5825fe60fd
samza-core/src/main/scala/org/apache/samza/config/TaskConfig.scala 0c742d83c2f60d2448a79376677713a1ff0b11ec
samza-core/src/main/scala/org/apache/samza/config/WrappedChooserConfig.scala PRE-CREATION
samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala 62bd243115e71612e00784124baa972b33e56cb7
samza-core/src/main/scala/org/apache/samza/system/DefaultChooser.scala 5a72e7a3bfba0f06a5a98c6ba26865800d7780b9
samza-core/src/main/scala/org/apache/samza/system/SystemConsumers.scala d24671ec64a42ede6f779effe9c845e1cbbc5e51
samza-core/src/main/scala/org/apache/samza/system/chooser/BatchingChooser.scala PRE-CREATION
samza-core/src/main/scala/org/apache/samza/system/chooser/BootstrappingChooser.scala PRE-CREATION
samza-core/src/main/scala/org/apache/samza/system/chooser/RoundRobinChooser.scala PRE-CREATION
samza-core/src/main/scala/org/apache/samza/system/chooser/TieredPriorityChooser.scala PRE-CREATION
samza-core/src/main/scala/org/apache/samza/system/chooser/WrappedChooser.scala PRE-CREATION
samza-core/src/test/scala/org/apache/samza/system/TestSystemConsumers.scala PRE-CREATION
samza-core/src/test/scala/org/apache/samza/system/chooser/MockMessageChooser.scala PRE-CREATION
samza-core/src/test/scala/org/apache/samza/system/chooser/TestBatchingChooser.scala PRE-CREATION
samza-core/src/test/scala/org/apache/samza/system/chooser/TestBootstrappingChooser.scala PRE-CREATION
samza-core/src/test/scala/org/apache/samza/system/chooser/TestRoundRobinChooser.scala PRE-CREATION
samza-core/src/test/scala/org/apache/samza/system/chooser/TestTieredPriorityChooser.scala PRE-CREATION
samza-core/src/test/scala/org/apache/samza/system/chooser/TestWrappedChooser.scala PRE-CREATION
samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemAdmin.scala 183c6ccce39dedaef9dba56d5b61ffdedfc9d08a
samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemFactory.scala ba08af85892f0793d01e3118fa7ad6d569a0f238
samza-test/src/test/scala/org/apache/samza/test/integration/TestStatefulTask.scala 7d4e20a8bdc7a45b0a1b464a6f4b868d1d03eab0
samza-yarn/src/test/scala/org/apache/samza/job/yarn/TestSamzaAppMasterTaskManager.scala 68050528cd5f8acfe3a1f7563b4e7fe6c7473be5
Diff: https://reviews.apache.org/r/13725/diff/
Testing
-------
Thanks,
Chris Riccomini
Re: Review Request 13725: SAMZA-2
Posted by Chris Riccomini <cr...@apache.org>.
-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/13725/
-----------------------------------------------------------
(Updated Oct. 1, 2013, 9:48 p.m.)
Review request for samza.
Repository: samza
Description (updated)
-------
use the systemAdmins from samza container in wrapped chooser, rather than doing wiring internally.
changing ordering of wrapped chooser from bootstrap -> priority -> batching to bootstrap -> batching -> priority.
switch default chooser to WrappedChooser. Always use wrapped chooser. Remove task.chooser.wrapped.class.
wrote default chooser test that bootstraps, prioritizes, and batches.
turning on priority tests for default chooser. refactor default chooser a bit to make it easier to test.
test default chooser with just round robin.
bug fix to use round robin chooser factory in default chooser
change wiring for default chooser to make it more testable. add start/stop/register tests to all choosers.
add one more test in priority chooser.
add unit test for tiered priority chooser
add unit test for bootstrapping chooser.
turn default chooser on in samza container. add license to test stateful task. add more docs to default chooser. clean up batching unit test.
add unit test for batching chooser.
only wire in wrapping message choosers when we need them. add docs to the default chooser factory.
rename class to BootstrappingChooser.
refactor to move into org.apache.samza.system.chooser.
build latest message offset map.
add a streams-behind-chooser that guarantees one message from each SystemStream before choose is called.
add start/stop/register back. all tests pass.
fix bug -- should allow manual override if Int.MaxInt for bootstrap streams.
minor bug in default chooser. was re-using same chooser everywhere.
adding wiring in default chooser.
initial pass adding composed message choosers.
rebase to master, which includes SAMZA-25 metrics. fix several tests that were broken after removing start/stop/register.
adding more docs for round robin
remove start/stop/register
cleanup some wiki markdown in MessageChooser javadoc.
add more javadocs to the message chooser.
Merge branch 'SAMZA-2_fine-grain-control-over-stream-consumption' of github.com:criccomini/incubator-samza into SAMZA-2_fine-grain-control-over-stream-consumption
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 (updated)
-----
docs/learn/documentation/0.7.0/container/streams.md b15c34d3f68e32d56e0da8af91d78e47a5110f67
samza-api/src/main/java/org/apache/samza/system/MessageChooser.java 306b2902303c72f3d7a3eb313f55d7e88d21e00d
samza-api/src/main/java/org/apache/samza/system/SystemAdmin.java c902d414484e05ae75c9ca58ad9629cb01120f62
samza-api/src/main/java/org/apache/samza/system/chooser/MessageChooser.java PRE-CREATION
samza-api/src/main/java/org/apache/samza/util/SinglePartitionSystemAdmin.java e4ed30bf363142d82a3c40909e160b5825fe60fd
samza-core/src/main/scala/org/apache/samza/config/TaskConfig.scala 0c742d83c2f60d2448a79376677713a1ff0b11ec
samza-core/src/main/scala/org/apache/samza/config/WrappedChooserConfig.scala PRE-CREATION
samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala 62bd243115e71612e00784124baa972b33e56cb7
samza-core/src/main/scala/org/apache/samza/system/DefaultChooser.scala 5a72e7a3bfba0f06a5a98c6ba26865800d7780b9
samza-core/src/main/scala/org/apache/samza/system/SystemConsumers.scala d24671ec64a42ede6f779effe9c845e1cbbc5e51
samza-core/src/main/scala/org/apache/samza/system/chooser/BatchingChooser.scala PRE-CREATION
samza-core/src/main/scala/org/apache/samza/system/chooser/BootstrappingChooser.scala PRE-CREATION
samza-core/src/main/scala/org/apache/samza/system/chooser/RoundRobinChooser.scala PRE-CREATION
samza-core/src/main/scala/org/apache/samza/system/chooser/TieredPriorityChooser.scala PRE-CREATION
samza-core/src/main/scala/org/apache/samza/system/chooser/WrappedChooser.scala PRE-CREATION
samza-core/src/test/scala/org/apache/samza/system/TestSystemConsumers.scala PRE-CREATION
samza-core/src/test/scala/org/apache/samza/system/chooser/MockMessageChooser.scala PRE-CREATION
samza-core/src/test/scala/org/apache/samza/system/chooser/TestBatchingChooser.scala PRE-CREATION
samza-core/src/test/scala/org/apache/samza/system/chooser/TestBootstrappingChooser.scala PRE-CREATION
samza-core/src/test/scala/org/apache/samza/system/chooser/TestRoundRobinChooser.scala PRE-CREATION
samza-core/src/test/scala/org/apache/samza/system/chooser/TestTieredPriorityChooser.scala PRE-CREATION
samza-core/src/test/scala/org/apache/samza/system/chooser/TestWrappedChooser.scala PRE-CREATION
samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemAdmin.scala 183c6ccce39dedaef9dba56d5b61ffdedfc9d08a
samza-test/src/test/scala/org/apache/samza/test/integration/TestStatefulTask.scala 7d4e20a8bdc7a45b0a1b464a6f4b868d1d03eab0
samza-yarn/src/test/scala/org/apache/samza/job/yarn/TestSamzaAppMasterTaskManager.scala 68050528cd5f8acfe3a1f7563b4e7fe6c7473be5
Diff: https://reviews.apache.org/r/13725/diff/
Testing
-------
Thanks,
Chris Riccomini
Re: Review Request 13725: SAMZA-2
Posted by Sriram Subramanian <sr...@gmail.com>.
-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/13725/#review26548
-----------------------------------------------------------
1. DefaultChooser should really be default. We should not be allowing anything to be configurable within in. The "task.chooser.wrapped.class" is hard to understand. This should just be hard coded to RoundRobin.
2. BatchingChooser does not seem to work with TieredPriorityChooser. Assume there were three priorities 3 , 2 , 1 and each priority has a Batch chooser. When choose gets called the first time assume there were no data in Priority 3 tier. We would then call choose on priority 2 chooser. In the next loop, if there were new data in Priority 3 tier, we would start choosing that. I think we should be able to use PriorityChooser instead of TierPriorityChooser and instead of composing BatchChooser inside TierPriorityChooser, I would compose BatchChooser to compose a BootstrapChooser which composes a PriorityChooser. Events of the same priority are decided using the PriorityQueues compare method.
- Sriram Subramanian
On Sept. 26, 2013, 3:03 a.m., Chris Riccomini wrote:
>
> -----------------------------------------------------------
> This is an automatically generated e-mail. To reply, visit:
> https://reviews.apache.org/r/13725/
> -----------------------------------------------------------
>
> (Updated Sept. 26, 2013, 3:03 a.m.)
>
>
> Review request for samza.
>
>
> Repository: samza
>
>
> Description
> -------
>
> wrote default chooser test that bootstraps, prioritizes, and batches.
>
>
> turning on priority tests for default chooser. refactor default chooser a bit to make it easier to test.
>
>
> test default chooser with just round robin.
>
>
> bug fix to use round robin chooser factory in default chooser
>
>
> change wiring for default chooser to make it more testable. add start/stop/register tests to all choosers.
>
>
> add one more test in priority chooser.
>
>
> add unit test for tiered priority chooser
>
>
> add unit test for bootstrapping chooser.
>
>
> turn default chooser on in samza container. add license to test stateful task. add more docs to default chooser. clean up batching unit test.
>
>
> add unit test for batching chooser.
>
>
> only wire in wrapping message choosers when we need them. add docs to the default chooser factory.
>
>
> rename class to BootstrappingChooser.
>
>
> refactor to move into org.apache.samza.system.chooser.
>
>
> build latest message offset map.
>
>
> add a streams-behind-chooser that guarantees one message from each SystemStream before choose is called.
>
>
> add start/stop/register back. all tests pass.
>
>
> fix bug -- should allow manual override if Int.MaxInt for bootstrap streams.
>
>
> minor bug in default chooser. was re-using same chooser everywhere.
>
>
> adding wiring in default chooser.
>
>
> initial pass adding composed message choosers.
>
>
> rebase to master, which includes SAMZA-25 metrics. fix several tests that were broken after removing start/stop/register.
>
>
> adding more docs for round robin
>
>
> remove start/stop/register
>
>
> cleanup some wiki markdown in MessageChooser javadoc.
>
>
> add more javadocs to the message chooser.
>
>
> Merge branch 'SAMZA-2_fine-grain-control-over-stream-consumption' of github.com:criccomini/incubator-samza into SAMZA-2_fine-grain-control-over-stream-consumption
>
>
> 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 b15c34d3f68e32d56e0da8af91d78e47a5110f67
> samza-api/src/main/java/org/apache/samza/system/MessageChooser.java 306b2902303c72f3d7a3eb313f55d7e88d21e00d
> samza-api/src/main/java/org/apache/samza/system/SystemAdmin.java c902d414484e05ae75c9ca58ad9629cb01120f62
> samza-api/src/main/java/org/apache/samza/system/chooser/MessageChooser.java PRE-CREATION
> samza-api/src/main/java/org/apache/samza/system/chooser/PriorityChooser.java PRE-CREATION
> samza-api/src/main/java/org/apache/samza/util/SinglePartitionSystemAdmin.java e4ed30bf363142d82a3c40909e160b5825fe60fd
> samza-api/src/test/java/org/apache/samza/system/chooser/TestPriorityChooser.java PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/config/DefaultChooserConfig.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 62bd243115e71612e00784124baa972b33e56cb7
> samza-core/src/main/scala/org/apache/samza/system/DefaultChooser.scala 5a72e7a3bfba0f06a5a98c6ba26865800d7780b9
> samza-core/src/main/scala/org/apache/samza/system/SystemConsumers.scala d24671ec64a42ede6f779effe9c845e1cbbc5e51
> samza-core/src/main/scala/org/apache/samza/system/chooser/BatchingChooser.scala PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/system/chooser/BootstrappingChooser.scala PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/system/chooser/DefaultChooser.scala PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/system/chooser/RoundRobinChooser.scala PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/system/chooser/TieredPriorityChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/TestSystemConsumers.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/MockMessageChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestBatchingChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestBootstrappingChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestDefaultChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestRoundRobinChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestTieredPriorityChooser.scala PRE-CREATION
> samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemAdmin.scala 183c6ccce39dedaef9dba56d5b61ffdedfc9d08a
> samza-test/src/test/scala/org/apache/samza/test/integration/TestStatefulTask.scala 7d4e20a8bdc7a45b0a1b464a6f4b868d1d03eab0
> samza-yarn/src/test/scala/org/apache/samza/job/yarn/TestSamzaAppMasterTaskManager.scala 68050528cd5f8acfe3a1f7563b4e7fe6c7473be5
>
> Diff: https://reviews.apache.org/r/13725/diff/
>
>
> Testing
> -------
>
>
> Thanks,
>
> Chris Riccomini
>
>
Re: Review Request 13725: SAMZA-2
Posted by Chris Riccomini <cr...@apache.org>.
> On Oct. 1, 2013, 7:59 a.m., Sriram Subramanian wrote:
> > 1. DefaultChooser should really be default. We should not be allowing anything to be configurable within in. The "task.chooser.wrapped.class" is hard to understand. This should just be hard coded to RoundRobin.
> >
> > 2. BatchingChooser does not seem to work with TieredPriorityChooser. Assume there were three priorities 3 , 2 , 1 and each priority has a Batch chooser. When choose gets called the first time assume there were no data in Priority 3 tier. We would then call choose on priority 2 chooser. In the next loop, if there were new data in Priority 3 tier, we would start choosing that. I think we should be able to use PriorityChooser instead of TierPriorityChooser and instead of composing BatchChooser inside TierPriorityChooser, I would compose BatchChooser to compose a BootstrapChooser which composes a PriorityChooser. Events of the same priority are decided using the PriorityQueues compare method.
The behavior you describe in (2) was intentional. It's a philosophical question whether batching should override a higher priority message or not. Thinking on it more, I don't see much down side to the alternative behavior you describe, though, and it'll lead to less thrashing (batch interrupts) in cases where the higher priority stream is low volume. I'll make the change.
A second note on eliminating the tiered priority chooser. This is a little bit trickier if we want to do the composition that you describe in the next comment. In such a case, we'll be given something like RoundRobinChooser to break ties for the PriorityChooser (via task.chooser.class). It's unclear to me how to convert a compare(o1, o2) call to a RoundRobinChooser.choose call. The nice thing about the TieredPriorityChooser is that it does this simply by having one chooser for each priority tier. The priority chooser, on the other hand, is more useful if you want to build a chooser that prefers messages based by time (for example). The point I'm trying to convey is that these two choosers actually do subtly different things.
Off the top of my head, I can't really see a good way to use a MessageChooser inside of PriorityChooser's Comparator.compare method. I'll have to think about it, but I'm going to keep both if nothing comes to mind. If you have a concrete proposal on how to do this, let me know, and I'll make the changes.
- Chris
-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/13725/#review26549
-----------------------------------------------------------
On Sept. 26, 2013, 3:03 a.m., Chris Riccomini wrote:
>
> -----------------------------------------------------------
> This is an automatically generated e-mail. To reply, visit:
> https://reviews.apache.org/r/13725/
> -----------------------------------------------------------
>
> (Updated Sept. 26, 2013, 3:03 a.m.)
>
>
> Review request for samza.
>
>
> Repository: samza
>
>
> Description
> -------
>
> wrote default chooser test that bootstraps, prioritizes, and batches.
>
>
> turning on priority tests for default chooser. refactor default chooser a bit to make it easier to test.
>
>
> test default chooser with just round robin.
>
>
> bug fix to use round robin chooser factory in default chooser
>
>
> change wiring for default chooser to make it more testable. add start/stop/register tests to all choosers.
>
>
> add one more test in priority chooser.
>
>
> add unit test for tiered priority chooser
>
>
> add unit test for bootstrapping chooser.
>
>
> turn default chooser on in samza container. add license to test stateful task. add more docs to default chooser. clean up batching unit test.
>
>
> add unit test for batching chooser.
>
>
> only wire in wrapping message choosers when we need them. add docs to the default chooser factory.
>
>
> rename class to BootstrappingChooser.
>
>
> refactor to move into org.apache.samza.system.chooser.
>
>
> build latest message offset map.
>
>
> add a streams-behind-chooser that guarantees one message from each SystemStream before choose is called.
>
>
> add start/stop/register back. all tests pass.
>
>
> fix bug -- should allow manual override if Int.MaxInt for bootstrap streams.
>
>
> minor bug in default chooser. was re-using same chooser everywhere.
>
>
> adding wiring in default chooser.
>
>
> initial pass adding composed message choosers.
>
>
> rebase to master, which includes SAMZA-25 metrics. fix several tests that were broken after removing start/stop/register.
>
>
> adding more docs for round robin
>
>
> remove start/stop/register
>
>
> cleanup some wiki markdown in MessageChooser javadoc.
>
>
> add more javadocs to the message chooser.
>
>
> Merge branch 'SAMZA-2_fine-grain-control-over-stream-consumption' of github.com:criccomini/incubator-samza into SAMZA-2_fine-grain-control-over-stream-consumption
>
>
> 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 b15c34d3f68e32d56e0da8af91d78e47a5110f67
> samza-api/src/main/java/org/apache/samza/system/MessageChooser.java 306b2902303c72f3d7a3eb313f55d7e88d21e00d
> samza-api/src/main/java/org/apache/samza/system/SystemAdmin.java c902d414484e05ae75c9ca58ad9629cb01120f62
> samza-api/src/main/java/org/apache/samza/system/chooser/MessageChooser.java PRE-CREATION
> samza-api/src/main/java/org/apache/samza/system/chooser/PriorityChooser.java PRE-CREATION
> samza-api/src/main/java/org/apache/samza/util/SinglePartitionSystemAdmin.java e4ed30bf363142d82a3c40909e160b5825fe60fd
> samza-api/src/test/java/org/apache/samza/system/chooser/TestPriorityChooser.java PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/config/DefaultChooserConfig.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 62bd243115e71612e00784124baa972b33e56cb7
> samza-core/src/main/scala/org/apache/samza/system/DefaultChooser.scala 5a72e7a3bfba0f06a5a98c6ba26865800d7780b9
> samza-core/src/main/scala/org/apache/samza/system/SystemConsumers.scala d24671ec64a42ede6f779effe9c845e1cbbc5e51
> samza-core/src/main/scala/org/apache/samza/system/chooser/BatchingChooser.scala PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/system/chooser/BootstrappingChooser.scala PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/system/chooser/DefaultChooser.scala PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/system/chooser/RoundRobinChooser.scala PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/system/chooser/TieredPriorityChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/TestSystemConsumers.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/MockMessageChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestBatchingChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestBootstrappingChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestDefaultChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestRoundRobinChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestTieredPriorityChooser.scala PRE-CREATION
> samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemAdmin.scala 183c6ccce39dedaef9dba56d5b61ffdedfc9d08a
> samza-test/src/test/scala/org/apache/samza/test/integration/TestStatefulTask.scala 7d4e20a8bdc7a45b0a1b464a6f4b868d1d03eab0
> samza-yarn/src/test/scala/org/apache/samza/job/yarn/TestSamzaAppMasterTaskManager.scala 68050528cd5f8acfe3a1f7563b4e7fe6c7473be5
>
> Diff: https://reviews.apache.org/r/13725/diff/
>
>
> Testing
> -------
>
>
> Thanks,
>
> Chris Riccomini
>
>
Re: Review Request 13725: SAMZA-2
Posted by Sriram Subramanian <sr...@gmail.com>.
-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/13725/#review26549
-----------------------------------------------------------
1. DefaultChooser should really be default. We should not be allowing anything to be configurable within in. The "task.chooser.wrapped.class" is hard to understand. This should just be hard coded to RoundRobin.
2. BatchingChooser does not seem to work with TieredPriorityChooser. Assume there were three priorities 3 , 2 , 1 and each priority has a Batch chooser. When choose gets called the first time assume there were no data in Priority 3 tier. We would then call choose on priority 2 chooser. In the next loop, if there were new data in Priority 3 tier, we would start choosing that. I think we should be able to use PriorityChooser instead of TierPriorityChooser and instead of composing BatchChooser inside TierPriorityChooser, I would compose BatchChooser to compose a BootstrapChooser which composes a PriorityChooser. Events of the same priority are decided using the PriorityQueues compare method.
- Sriram Subramanian
On Sept. 26, 2013, 3:03 a.m., Chris Riccomini wrote:
>
> -----------------------------------------------------------
> This is an automatically generated e-mail. To reply, visit:
> https://reviews.apache.org/r/13725/
> -----------------------------------------------------------
>
> (Updated Sept. 26, 2013, 3:03 a.m.)
>
>
> Review request for samza.
>
>
> Repository: samza
>
>
> Description
> -------
>
> wrote default chooser test that bootstraps, prioritizes, and batches.
>
>
> turning on priority tests for default chooser. refactor default chooser a bit to make it easier to test.
>
>
> test default chooser with just round robin.
>
>
> bug fix to use round robin chooser factory in default chooser
>
>
> change wiring for default chooser to make it more testable. add start/stop/register tests to all choosers.
>
>
> add one more test in priority chooser.
>
>
> add unit test for tiered priority chooser
>
>
> add unit test for bootstrapping chooser.
>
>
> turn default chooser on in samza container. add license to test stateful task. add more docs to default chooser. clean up batching unit test.
>
>
> add unit test for batching chooser.
>
>
> only wire in wrapping message choosers when we need them. add docs to the default chooser factory.
>
>
> rename class to BootstrappingChooser.
>
>
> refactor to move into org.apache.samza.system.chooser.
>
>
> build latest message offset map.
>
>
> add a streams-behind-chooser that guarantees one message from each SystemStream before choose is called.
>
>
> add start/stop/register back. all tests pass.
>
>
> fix bug -- should allow manual override if Int.MaxInt for bootstrap streams.
>
>
> minor bug in default chooser. was re-using same chooser everywhere.
>
>
> adding wiring in default chooser.
>
>
> initial pass adding composed message choosers.
>
>
> rebase to master, which includes SAMZA-25 metrics. fix several tests that were broken after removing start/stop/register.
>
>
> adding more docs for round robin
>
>
> remove start/stop/register
>
>
> cleanup some wiki markdown in MessageChooser javadoc.
>
>
> add more javadocs to the message chooser.
>
>
> Merge branch 'SAMZA-2_fine-grain-control-over-stream-consumption' of github.com:criccomini/incubator-samza into SAMZA-2_fine-grain-control-over-stream-consumption
>
>
> 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 b15c34d3f68e32d56e0da8af91d78e47a5110f67
> samza-api/src/main/java/org/apache/samza/system/MessageChooser.java 306b2902303c72f3d7a3eb313f55d7e88d21e00d
> samza-api/src/main/java/org/apache/samza/system/SystemAdmin.java c902d414484e05ae75c9ca58ad9629cb01120f62
> samza-api/src/main/java/org/apache/samza/system/chooser/MessageChooser.java PRE-CREATION
> samza-api/src/main/java/org/apache/samza/system/chooser/PriorityChooser.java PRE-CREATION
> samza-api/src/main/java/org/apache/samza/util/SinglePartitionSystemAdmin.java e4ed30bf363142d82a3c40909e160b5825fe60fd
> samza-api/src/test/java/org/apache/samza/system/chooser/TestPriorityChooser.java PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/config/DefaultChooserConfig.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 62bd243115e71612e00784124baa972b33e56cb7
> samza-core/src/main/scala/org/apache/samza/system/DefaultChooser.scala 5a72e7a3bfba0f06a5a98c6ba26865800d7780b9
> samza-core/src/main/scala/org/apache/samza/system/SystemConsumers.scala d24671ec64a42ede6f779effe9c845e1cbbc5e51
> samza-core/src/main/scala/org/apache/samza/system/chooser/BatchingChooser.scala PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/system/chooser/BootstrappingChooser.scala PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/system/chooser/DefaultChooser.scala PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/system/chooser/RoundRobinChooser.scala PRE-CREATION
> samza-core/src/main/scala/org/apache/samza/system/chooser/TieredPriorityChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/TestSystemConsumers.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/MockMessageChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestBatchingChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestBootstrappingChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestDefaultChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestRoundRobinChooser.scala PRE-CREATION
> samza-core/src/test/scala/org/apache/samza/system/chooser/TestTieredPriorityChooser.scala PRE-CREATION
> samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemAdmin.scala 183c6ccce39dedaef9dba56d5b61ffdedfc9d08a
> samza-test/src/test/scala/org/apache/samza/test/integration/TestStatefulTask.scala 7d4e20a8bdc7a45b0a1b464a6f4b868d1d03eab0
> samza-yarn/src/test/scala/org/apache/samza/job/yarn/TestSamzaAppMasterTaskManager.scala 68050528cd5f8acfe3a1f7563b4e7fe6c7473be5
>
> Diff: https://reviews.apache.org/r/13725/diff/
>
>
> Testing
> -------
>
>
> Thanks,
>
> Chris Riccomini
>
>
Re: Review Request 13725: SAMZA-2
Posted by Chris Riccomini <cr...@apache.org>.
-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/13725/
-----------------------------------------------------------
(Updated Sept. 26, 2013, 3:03 a.m.)
Review request for samza.
Repository: samza
Description (updated)
-------
wrote default chooser test that bootstraps, prioritizes, and batches.
turning on priority tests for default chooser. refactor default chooser a bit to make it easier to test.
test default chooser with just round robin.
bug fix to use round robin chooser factory in default chooser
change wiring for default chooser to make it more testable. add start/stop/register tests to all choosers.
add one more test in priority chooser.
add unit test for tiered priority chooser
add unit test for bootstrapping chooser.
turn default chooser on in samza container. add license to test stateful task. add more docs to default chooser. clean up batching unit test.
add unit test for batching chooser.
only wire in wrapping message choosers when we need them. add docs to the default chooser factory.
rename class to BootstrappingChooser.
refactor to move into org.apache.samza.system.chooser.
build latest message offset map.
add a streams-behind-chooser that guarantees one message from each SystemStream before choose is called.
add start/stop/register back. all tests pass.
fix bug -- should allow manual override if Int.MaxInt for bootstrap streams.
minor bug in default chooser. was re-using same chooser everywhere.
adding wiring in default chooser.
initial pass adding composed message choosers.
rebase to master, which includes SAMZA-25 metrics. fix several tests that were broken after removing start/stop/register.
adding more docs for round robin
remove start/stop/register
cleanup some wiki markdown in MessageChooser javadoc.
add more javadocs to the message chooser.
Merge branch 'SAMZA-2_fine-grain-control-over-stream-consumption' of github.com:criccomini/incubator-samza into SAMZA-2_fine-grain-control-over-stream-consumption
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 (updated)
-----
docs/learn/documentation/0.7.0/container/streams.md b15c34d3f68e32d56e0da8af91d78e47a5110f67
samza-api/src/main/java/org/apache/samza/system/MessageChooser.java 306b2902303c72f3d7a3eb313f55d7e88d21e00d
samza-api/src/main/java/org/apache/samza/system/SystemAdmin.java c902d414484e05ae75c9ca58ad9629cb01120f62
samza-api/src/main/java/org/apache/samza/system/chooser/MessageChooser.java PRE-CREATION
samza-api/src/main/java/org/apache/samza/system/chooser/PriorityChooser.java PRE-CREATION
samza-api/src/main/java/org/apache/samza/util/SinglePartitionSystemAdmin.java e4ed30bf363142d82a3c40909e160b5825fe60fd
samza-api/src/test/java/org/apache/samza/system/chooser/TestPriorityChooser.java PRE-CREATION
samza-core/src/main/scala/org/apache/samza/config/DefaultChooserConfig.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 62bd243115e71612e00784124baa972b33e56cb7
samza-core/src/main/scala/org/apache/samza/system/DefaultChooser.scala 5a72e7a3bfba0f06a5a98c6ba26865800d7780b9
samza-core/src/main/scala/org/apache/samza/system/SystemConsumers.scala d24671ec64a42ede6f779effe9c845e1cbbc5e51
samza-core/src/main/scala/org/apache/samza/system/chooser/BatchingChooser.scala PRE-CREATION
samza-core/src/main/scala/org/apache/samza/system/chooser/BootstrappingChooser.scala PRE-CREATION
samza-core/src/main/scala/org/apache/samza/system/chooser/DefaultChooser.scala PRE-CREATION
samza-core/src/main/scala/org/apache/samza/system/chooser/RoundRobinChooser.scala PRE-CREATION
samza-core/src/main/scala/org/apache/samza/system/chooser/TieredPriorityChooser.scala PRE-CREATION
samza-core/src/test/scala/org/apache/samza/system/TestSystemConsumers.scala PRE-CREATION
samza-core/src/test/scala/org/apache/samza/system/chooser/MockMessageChooser.scala PRE-CREATION
samza-core/src/test/scala/org/apache/samza/system/chooser/TestBatchingChooser.scala PRE-CREATION
samza-core/src/test/scala/org/apache/samza/system/chooser/TestBootstrappingChooser.scala PRE-CREATION
samza-core/src/test/scala/org/apache/samza/system/chooser/TestDefaultChooser.scala PRE-CREATION
samza-core/src/test/scala/org/apache/samza/system/chooser/TestRoundRobinChooser.scala PRE-CREATION
samza-core/src/test/scala/org/apache/samza/system/chooser/TestTieredPriorityChooser.scala PRE-CREATION
samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemAdmin.scala 183c6ccce39dedaef9dba56d5b61ffdedfc9d08a
samza-test/src/test/scala/org/apache/samza/test/integration/TestStatefulTask.scala 7d4e20a8bdc7a45b0a1b464a6f4b868d1d03eab0
samza-yarn/src/test/scala/org/apache/samza/job/yarn/TestSamzaAppMasterTaskManager.scala 68050528cd5f8acfe3a1f7563b4e7fe6c7473be5
Diff: https://reviews.apache.org/r/13725/diff/
Testing
-------
Thanks,
Chris Riccomini
Re: Review Request 13725: SAMZA-2
Posted by Chris Riccomini <cr...@apache.org>.
> On Sept. 12, 2013, 9:52 p.m., Sriram Subramanian wrote:
> > 1. I am not sure if the statement below is true. The chooser's update method can be called multiple times for the same topic partition before choose is called.
> >
> > "A MessageChooser will receive, at most, one outstanding envelope per system/stream/partition combination. For example, if update is called for partition 7 of kafka.mystream, then update will not be called with an envelope from partition 7 of kafka.mystream until the previous envelope has been returned via the choose method."
> >
> > 2. Can you provide a usecase where register and start would be useful and what sort of computations can be done in these methods. All our choosers don't seem to do anything for these methods. I am trying to think what are the use cases this is useful Vs understandability of the api.
> >
> > 3. Could you add the assumptions made in the RoundRobin Scheduler about the container implementation.
1. Hmm. Can you point out where/how the update method can be called more than once for a single SSP? This is a bug.
2. No I can't. I'm backing off this interface change. Will revert.
3. Yes, will do.
- Chris
-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/13725/#review26064
-----------------------------------------------------------
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
>
>
Re: Review Request 13725: SAMZA-2
Posted by Sriram Subramanian <sr...@gmail.com>.
-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/13725/#review26064
-----------------------------------------------------------
1. I am not sure if the statement below is true. The chooser's update method can be called multiple times for the same topic partition before choose is called.
"A MessageChooser will receive, at most, one outstanding envelope per system/stream/partition combination. For example, if update is called for partition 7 of kafka.mystream, then update will not be called with an envelope from partition 7 of kafka.mystream until the previous envelope has been returned via the choose method."
2. Can you provide a usecase where register and start would be useful and what sort of computations can be done in these methods. All our choosers don't seem to do anything for these methods. I am trying to think what are the use cases this is useful Vs understandability of the api.
3. Could you add the assumptions made in the RoundRobin Scheduler about the container implementation.
- Sriram Subramanian
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
>
>
Re: Review Request 13725: SAMZA-2
Posted by Chris Riccomini <cr...@apache.org>.
-----------------------------------------------------------
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 (updated)
-------
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 (updated)
-----
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
Re: Review Request 13725: SAMZA-2
Posted by Chris Riccomini <cr...@apache.org>.
> On Aug. 28, 2013, 9:39 p.m., Sriram Subramanian wrote:
> > I don't really like the fact that the chooser's implementation depends on how it gets called. That seems fragile.
>
> Jay Kreps wrote:
> In fairness that contract is documented in the interface so changing it is effectively changing the interface. I think it is actually a safe assumption because allowing prioritization of more than one message from a given stream breaks the linear processing requirement of logs and offsets. That said, I think maybe at heart your complaint was similar to mine which is just that update/choose is an unusual way to express an ordering. But I can't think of a better one that is fully general (the priority thing was as close as we got and as you pointed out that straight-jackets you into a heap and isn't the most natural for other cases--e.g. doing round-robin by using a continually decremented counter is clever but weird, right?).
Three thoughts:
1. Do nothing.
2. Make the RoundRobinChooser protect itself (throw an exception), to guarantee only one message is received at a time.
3. Implement a RoundRobinChooser that works if multiple messages from a single SSP are update()'d even though none have been returned yet.
The third option is essentially dangerous and wrong. Doing this can result in re-ordering messages within a partition, in the general case. In the specific RoundRobinChooser case, it should be fine, since round robin'ing is just FIFO'ing within a single stream partition, and no re-ordering is taking place. Still, I don't like supporting something that is broken behavior on the part of the container.
I'm going to leave it how it is.
- Chris
-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/13725/#review25688
-----------------------------------------------------------
On Aug. 28, 2013, 6:23 p.m., Chris Riccomini wrote:
>
> -----------------------------------------------------------
> This is an automatically generated e-mail. To reply, visit:
> https://reviews.apache.org/r/13725/
> -----------------------------------------------------------
>
> (Updated Aug. 28, 2013, 6:23 p.m.)
>
>
> Review request for samza.
>
>
> Repository: samza
>
>
> Description
> -------
>
> 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/MessageChooserFactory.java PRE-CREATION
> 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/StreamChooser.scala PRE-CREATION
> 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
>
> Diff: https://reviews.apache.org/r/13725/diff/
>
>
> Testing
> -------
>
>
> Thanks,
>
> Chris Riccomini
>
>
Re: Review Request 13725: SAMZA-2
Posted by Jay Kreps <bo...@gmail.com>.
> On Aug. 28, 2013, 9:39 p.m., Sriram Subramanian wrote:
> > I don't really like the fact that the chooser's implementation depends on how it gets called. That seems fragile.
In fairness that contract is documented in the interface so changing it is effectively changing the interface. I think it is actually a safe assumption because allowing prioritization of more than one message from a given stream breaks the linear processing requirement of logs and offsets. That said, I think maybe at heart your complaint was similar to mine which is just that update/choose is an unusual way to express an ordering. But I can't think of a better one that is fully general (the priority thing was as close as we got and as you pointed out that straight-jackets you into a heap and isn't the most natural for other cases--e.g. doing round-robin by using a continually decremented counter is clever but weird, right?).
- Jay
-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/13725/#review25688
-----------------------------------------------------------
On Aug. 28, 2013, 6:23 p.m., Chris Riccomini wrote:
>
> -----------------------------------------------------------
> This is an automatically generated e-mail. To reply, visit:
> https://reviews.apache.org/r/13725/
> -----------------------------------------------------------
>
> (Updated Aug. 28, 2013, 6:23 p.m.)
>
>
> Review request for samza.
>
>
> Repository: samza
>
>
> Description
> -------
>
> 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/MessageChooserFactory.java PRE-CREATION
> 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/StreamChooser.scala PRE-CREATION
> 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
>
> Diff: https://reviews.apache.org/r/13725/diff/
>
>
> Testing
> -------
>
>
> Thanks,
>
> Chris Riccomini
>
>
Re: Review Request 13725: SAMZA-2
Posted by Sriram Subramanian <sr...@gmail.com>.
-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/13725/#review25688
-----------------------------------------------------------
I don't really like the fact that the chooser's implementation depends on how it gets called. That seems fragile.
- Sriram Subramanian
On Aug. 28, 2013, 6:23 p.m., Chris Riccomini wrote:
>
> -----------------------------------------------------------
> This is an automatically generated e-mail. To reply, visit:
> https://reviews.apache.org/r/13725/
> -----------------------------------------------------------
>
> (Updated Aug. 28, 2013, 6:23 p.m.)
>
>
> Review request for samza.
>
>
> Repository: samza
>
>
> Description
> -------
>
> 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/MessageChooserFactory.java PRE-CREATION
> 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/StreamChooser.scala PRE-CREATION
> 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
>
> Diff: https://reviews.apache.org/r/13725/diff/
>
>
> Testing
> -------
>
>
> Thanks,
>
> Chris Riccomini
>
>
Re: Review Request 13725: SAMZA-2
Posted by Chris Riccomini <cr...@apache.org>.
> On Aug. 28, 2013, 8:42 p.m., Sriram Subramanian wrote:
> > I am confused how the RoundRobin implementation is different from a FIFO implementation. Does this depend on an implementation detail outside it?
Hey Sriram,
Yea, it depends on an outside detail, which is that update() will be called at most once per-system/stream/partition combination. This means that if you're given a message for kafka.mystream partition 7, you won't get another message from that SSP until you've returned the last one given to you (as a return value in the choose method). This behavior means that queueing messages at the end of the queue should result in a round-robin behavior, I think.
- Chris
-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/13725/#review25686
-----------------------------------------------------------
On Aug. 28, 2013, 6:23 p.m., Chris Riccomini wrote:
>
> -----------------------------------------------------------
> This is an automatically generated e-mail. To reply, visit:
> https://reviews.apache.org/r/13725/
> -----------------------------------------------------------
>
> (Updated Aug. 28, 2013, 6:23 p.m.)
>
>
> Review request for samza.
>
>
> Repository: samza
>
>
> Description
> -------
>
> 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/MessageChooserFactory.java PRE-CREATION
> 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/StreamChooser.scala PRE-CREATION
> 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
>
> Diff: https://reviews.apache.org/r/13725/diff/
>
>
> Testing
> -------
>
>
> Thanks,
>
> Chris Riccomini
>
>
Re: Review Request 13725: SAMZA-2
Posted by Sriram Subramanian <sr...@gmail.com>.
-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/13725/#review25686
-----------------------------------------------------------
I am confused how the RoundRobin implementation is different from a FIFO implementation. Does this depend on an implementation detail outside it?
- Sriram Subramanian
On Aug. 28, 2013, 6:23 p.m., Chris Riccomini wrote:
>
> -----------------------------------------------------------
> This is an automatically generated e-mail. To reply, visit:
> https://reviews.apache.org/r/13725/
> -----------------------------------------------------------
>
> (Updated Aug. 28, 2013, 6:23 p.m.)
>
>
> Review request for samza.
>
>
> Repository: samza
>
>
> Description
> -------
>
> 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/MessageChooserFactory.java PRE-CREATION
> 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/StreamChooser.scala PRE-CREATION
> 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
>
> Diff: https://reviews.apache.org/r/13725/diff/
>
>
> Testing
> -------
>
>
> Thanks,
>
> Chris Riccomini
>
>