Re: Producer stopped during leader switch

[David Yu]

That's good news. Thanks.

On Mon, Oct 31, 2016 at 9:59 PM Jagadish Venkatraman 
wrote:

> Hey David,
>
> Apologies for the delayed response. There are a bunch of producer *send*
> issues that were fixed in samza-11 (SAMZA-1028, SAMZA-1003). With Samza-11,
> there's improved resiliency and if retries fail, the exception will be
> propagated to your *StreamTask*, and will end up failing the container (if
> you don't swallow it).
>
> Thanks,
> Jagadish
>
>
>
> On Fri, Oct 28, 2016 at 10:12 AM, David Yu 
> wrote:
>
> > Hi,
> >
> > We recently experienced a Kafka broker crash. When a new broker was
> brought
> > up, we started seeing the following errors in Samza (0.10.1):
> >
> > WARN  o.a.k.c.producer.internals.Sender - Got error produce response with
> > correlation id 5199601 on topic-partition
> > session_key_partitioned_sessions-39, retrying (2147483646
> <(214)%20748-3646> attempts left).
> > Error: NOT_LEADER_FOR_PARTITION
> >
> > Is the Producer not able to detect the new broker/leader for that
> > partition?
> >
> > Thanks,
> > David
> >
>
>
>
> --
> Jagadish V,
> Graduate Student,
> Department of Computer Science,
> Stanford University
>

Re: Review Request 53326: SAMZA-1045 Move classes from samza-operator/api into samza-api.

[Fred Ji]

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


Ship it!




Ship It!

- Fred Ji


On Nov. 1, 2016, 6:40 a.m., Jagadish Venkatraman wrote:
> 
> ---
> This is an automatically generated e-mail. To reply, visit:
> https://reviews.apache.org/r/53326/
> ---
> 
> (Updated Nov. 1, 2016, 6:40 a.m.)
> 
> 
> Review request for samza, Jake Maes, Yi Pan (Data Infrastructure), Prateek 
> Maheshwari, and Xinyu Liu.
> 
> 
> Repository: samza
> 
> 
> Description
> ---
> 
> Refactor API classes from Samza-Operator/operator into samza-api. 
> 
> - Programmers write code against classes on samza-api, and it would be nice 
> to not have a dependency on samza-operator (when defining the transform logic 
> for the operator). 
> - For split-deployment to be effective, it'd be nice to independently evolve 
> samza-operator components.
> - Fixed checkstyle issues and formating.
> 
> Note: There are a couple of bugs that I found in the operator logic 
> (SAMZA-914/915). They are intentionally kept out of the scope this RB
> 
> 
> Diffs
> -
> 
>   build.gradle aeefd1cfd049efa37d20e12e34232600ae6bd127 
>   checkstyle/checkstyle.xml 770b5e7f7a091198bbf53b3908600f9ac0caa4c7 
>   gradle.properties 16e1f5d43f0415c511689480f8cb67d84e2baadf 
>   gradle/wrapper/gradle-wrapper.jar a7634b071cb255e91a4572934e55b8cd8877b3e4 
>   gradle/wrapper/gradle-wrapper.properties 
> 78596c0ebdd585e2d674cbcec930d0a8a2a08e74 
>   gradlew 91a7e269e19dfc62e27137a0b57ef3e430cee4fd 
>   samza-api/src/main/java/org/apache/samza/storage/StorageEngineFactory.java 
> 963ccf2b6222f0fee00705923d921f91ed481fbc 
>   
> samza-core/src/test/java/org/apache/samza/coordinator/stream/MockCoordinatorStreamWrappedConsumer.java
>  429573b480112c7491303dc410d78f37a308c4a7 
>   
> samza-core/src/test/java/org/apache/samza/coordinator/stream/TestCoordinatorStreamSystemConsumer.java
>  0e73e18bd55e343e1a5122be7e8f3c666b797dc5 
>   
> samza-operator/src/main/java/org/apache/samza/operators/api/MessageStream.java
>  b5e1028662a67e6248722ef7c842c565fef7a458 
>   
> samza-operator/src/main/java/org/apache/samza/operators/api/MessageStreams.java
>  59dd91c2f537828640b1ede6b6366e37f6b5c63b 
>   
> samza-operator/src/main/java/org/apache/samza/operators/api/TriggerBuilder.java
>  fc3ea37563ded2fcceeb332b0edfef45208bb9bb 
>   
> samza-operator/src/main/java/org/apache/samza/operators/api/WindowState.java 
> 402cc42f63bad0c8745c6a480b03769f92053622 
>   samza-operator/src/main/java/org/apache/samza/operators/api/Windows.java 
> e557b34c25ff425385aedae24bad6bb89ee03a30 
>   
> samza-operator/src/main/java/org/apache/samza/operators/api/data/IncomingSystemMessage.java
>  ba74618961d3ea53ad455de3263c59c0ca2a0fa7 
>   
> samza-operator/src/main/java/org/apache/samza/operators/api/data/InputSystemMessage.java
>  c7860254e38a897e42e00b60111526db0f7510ea 
>   
> samza-operator/src/main/java/org/apache/samza/operators/api/data/LongOffset.java
>  f059b337e299aab07c85af9866371c87fcf59786 
>   
> samza-operator/src/main/java/org/apache/samza/operators/api/data/Message.java 
> 9b53b4582ed12fa7d9948ff756ed9d0a4b38e280 
>   
> samza-operator/src/main/java/org/apache/samza/operators/api/data/Offset.java 
> 0fac2c0123b133566630bdf11e9b1c00a207bd2e 
>   
> samza-operator/src/main/java/org/apache/samza/operators/api/internal/Operators.java
>  e9bfe0b38c1048ecc7fe634ed7b0d71fee6f2ac6 
>   
> samza-operator/src/main/java/org/apache/samza/operators/api/internal/Trigger.java
>  33a0134e88b966d8f08658f18ca73564702305fa 
>   
> samza-operator/src/main/java/org/apache/samza/operators/api/internal/WindowFn.java
>  1fd88e76119960d1e6b83189bb571e64ac329489 
>   
> samza-operator/src/main/java/org/apache/samza/operators/api/internal/WindowOutput.java
>  e202c20cb0af6e53268e0c80820bfb1d7fc80892 
>   
> samza-operator/src/main/java/org/apache/samza/operators/impl/ChainedOperators.java
>  59de16bd11fee34b23baac14d1d32cf9df555058 
>   
> samza-operator/src/main/java/org/apache/samza/operators/impl/OperatorFactory.java
>  f16cbc6bf405851488c22b943a2256864b7a9f07 
>   
> samza-operator/src/main/java/org/apache/samza/operators/impl/OperatorImpl.java
>  3ca8bdeb7d73c72f469fa33a152ed7518f9e8ac4 
>   
> samza-operator/src/main/java/org/apache/samza/operators/impl/ProcessorContext.java
>  5a375bc7bb3b00fd252c6b1bbe211de8feddd942 
>   
> samza-operator/src/main/java/org/apache/samza/operators/impl/SimpleOperatorImpl.java
>  b29d9c8c412c70820ac8355137df447d373af1f6 
>   
> samza-operator/src/main/java/org/apache/samza/operators/impl/SinkOperatorImpl.java
>  5d25cfa106424f45660a45434c4abffd242b8672 
>   
> samza-operator/src/main/java/org/apache/samza/operators/impl/StateStoreImpl.java
>  f573fd04e1aaf827d2ca52a5edca8f7060723aa0 
> 

Re: Review Request 52492: Adding monitor to clean up stale local stores of jobs/tasks.

[Boris Shkolnik]

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


Ship it!





samza-rest/src/main/java/org/apache/samza/monitor/LocalStoreMonitor.java (line 
96)


nit. just for readability please move 
'InetAddress.getLocalHost().getHostName()' to before the 'if'


- Boris Shkolnik


On Oct. 31, 2016, 11:04 p.m., Shanthoosh Venkataraman wrote:
> 
> ---
> This is an automatically generated e-mail. To reply, visit:
> https://reviews.apache.org/r/52492/
> ---
> 
> (Updated Oct. 31, 2016, 11:04 p.m.)
> 
> 
> Review request for samza.
> 
> 
> Repository: samza
> 
> 
> Description
> ---
> 
> This patch contains the samza-rest monitor that periodically cleans up the 
> stale local stores of dead jobs/tasks. It performs the store deletion in two 
> phases. Initially it deletes the offset file in the local task stores if the 
> following condition is true. ((jobIsNotRunning || preferedHost != nmHost) && 
> offsetFilelastModifiedTime is greater than deleteRetention). During the 
> subsequent run, it deletes the local task stores if it does not contain 
> offset file. Please refer to the design doc of SAMZA-656 
> (https://issues.apache.org/jira/secure/attachment/12828083/DESIGN-SAMZA-656.pdf)
>  for more details.
> 
> 
> Diffs
> -
> 
>   build.gradle 2bea27b75288d3103178bc3762b9556f6e69cdd1 
>   samza-rest/src/main/java/org/apache/samza/monitor/JobsClient.java 
> PRE-CREATION 
>   samza-rest/src/main/java/org/apache/samza/monitor/LocalStoreMonitor.java 
> PRE-CREATION 
>   
> samza-rest/src/main/java/org/apache/samza/monitor/LocalStoreMonitorConfig.java
>  PRE-CREATION 
>   
> samza-rest/src/main/java/org/apache/samza/monitor/LocalStoreMonitorFactory.java
>  PRE-CREATION 
>   
> samza-rest/src/main/java/org/apache/samza/rest/resources/ResourceConstants.java
>  PRE-CREATION 
>   
> samza-rest/src/test/java/org/apache/samza/monitor/TestLocalStoreMonitor.java 
> PRE-CREATION 
> 
> Diff: https://reviews.apache.org/r/52492/diff/
> 
> 
> Testing
> ---
> 
> Unit testing and manual testing are done to verify the functionality.
> 
> 
> Thanks,
> 
> Shanthoosh Venkataraman
> 
>

Re: Review Request 47835: SAMZA-914: Initial draft for Java programming APIs on operators supporting DAGs

[Xinyu Liu]

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


Ship it!




THe RB looks great! I posted a couple of comments for further discussion, not 
to block the progress.


samza-operator/src/main/java/org/apache/samza/operators/api/MessageStreams.java 
(line 45)


This might be discussed before: why we use systemstreampartition instead of 
systemstream as the source? I think having an abstraction layer of stream will 
be much more convenient to the users to user this api. For example, instead of 

sources.foreach(source => source.map(...).filter()...)// per ssp handling

The api can be:
SystemMessageStream source = souces.get("system x", "stream y");
source.map(...).filter()...

This might not be a big deal if we only got one input stream. Things could 
get messy if we have more. For example, if we want to join, it will be much 
convenient to specify:

SystemMessageStream source1 = souces.get("system x", "stream 1");
SystemMessageStream source2 = souces.get("system x", "stream 2");
source1.join(source2, joinFunc)

Otherwise the user needs to join each partion and repeat the same logic for 
each partition. I couldn't think a case that a ssp level api will be useful 
since most the current stream processors operates on streams, not partitions.



samza-operator/src/main/java/org/apache/samza/operators/api/internal/Trigger.java
 (line 33)


The defination of early and late triggers seems to be a little different 
from what google dataflow has today. In Tyler's blog (stream 102), early and 
late are relevant if there is a watermark, which indicates the completion of 
input stream regarding to an event time. Early triggers happen before the 
watermark to solve the watermark being too slow problem and late triggers 
happen after the watermark to solve the watermark being too fast problem (late 
arrivals). For us, since there is no notion of watermark so far, it seems the 
semantics of early/late triggers are not well defined. I cannot tell when the 
window is closed (early or late tigger?) or can late trigger happens before 
early trigger? Seems the early and late meanings are loosely defined without a 
reference. Happy to talk about this more.


- Xinyu Liu


On Oct. 4, 2016, 12:45 a.m., Yi Pan (Data Infrastructure) wrote:
> 
> ---
> This is an automatically generated e-mail. To reply, visit:
> https://reviews.apache.org/r/47835/
> ---
> 
> (Updated Oct. 4, 2016, 12:45 a.m.)
> 
> 
> Review request for samza, Boris Shkolnik, Chris Pettitt, Chinmay Soman, Jake 
> Maes, Navina Ramesh, Jagadish Venkatraman, and Xinyu Liu.
> 
> 
> Bugs: SAMZA-914
> https://issues.apache.org/jira/browse/SAMZA-914
> 
> 
> Repository: samza
> 
> 
> Description
> ---
> 
> SAMZA-914: initial draft of operator programming API. Design doc attached to 
> SAMZA-914: 
> https://issues.apache.org/jira/secure/attachment/12821524/SAMZA-914_%20operator%20Java%20programming%20API%20-%20Google%20Docs.pdf
> 
> 
> Diffs
> -
> 
>   build.gradle 16facbbf4dff378c561461786ff186bd9eed 
>   gradle/dependency-versions.gradle 52e25aa53a1edc85d478b48898621b26508ad4bb 
>   samza-api/src/test/java/org/apache/samza/config/TestConfig.java 
> 5d066c5867e9df9e94e60bde825dedf10703b399 
>   
> samza-operator/src/main/java/org/apache/samza/operators/api/MessageStream.java
>  PRE-CREATION 
>   
> samza-operator/src/main/java/org/apache/samza/operators/api/MessageStreams.java
>  PRE-CREATION 
>   
> samza-operator/src/main/java/org/apache/samza/operators/api/TriggerBuilder.java
>  PRE-CREATION 
>   
> samza-operator/src/main/java/org/apache/samza/operators/api/WindowState.java 
> PRE-CREATION 
>   samza-operator/src/main/java/org/apache/samza/operators/api/Windows.java 
> PRE-CREATION 
>   
> samza-operator/src/main/java/org/apache/samza/operators/api/data/Message.java 
> PRE-CREATION 
>   
> samza-operator/src/main/java/org/apache/samza/operators/api/internal/Operators.java
>  PRE-CREATION 
>   
> samza-operator/src/main/java/org/apache/samza/operators/api/internal/Trigger.java
>  PRE-CREATION 
>   
> samza-operator/src/main/java/org/apache/samza/operators/api/internal/WindowFn.java
>  PRE-CREATION 
>   
> samza-operator/src/main/java/org/apache/samza/operators/api/internal/WindowOutput.java
>  PRE-CREATION 
>   
> samza-operator/src/main/java/org/apache/samza/operators/impl/ChainedOperators.java
>  PRE-CREATION 
>   
> samza-operator/src/main/java/org/apache/samza/operators/impl/OperatorImpl.java
>  PRE-CREATION 
>   
> samza-operator/src/main/java/org/apache/samza/operators/impl/ProcessorContext.java
>  P