----------------------------------------------------------- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/48213/#review136338 -----------------------------------------------------------
Fix it, then Ship it! samza-hdfs/src/main/scala/org/apache/samza/system/hdfs/HdfsSystemProducer.scala (line 86) <https://reviews.apache.org/r/48213/#comment201357> I think you can drop this synchronized call. Otherwise the one inside the function is redundant. samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemProducer.scala (lines 64 - 65) <https://reviews.apache.org/r/48213/#comment201358> Why not use null to indicate the created state? In particular, why don't you null out the producer when it is no longer needed. Leaving a ref holds in memory all of the objects reachable from the producer whether they are needed or not. Hopefully close on the producer covers that, but the null approach is safer with no apparent downside. samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemProducer.scala (line 97) <https://reviews.apache.org/r/48213/#comment201359> I think I see why you were trying to not null out the producer above. However, this can still NPE if start has not been called. Instead, now that producer is volatile, why not grab its state at the beginning of this function. If it is null throw an error, otherwise you can use it (assuming that producer handles close correctly). Another benefit is that you reduce the number of volatile reads you need to make. - Chris Pettitt On June 3, 2016, 10:09 p.m., Xinyu Liu wrote: > > ----------------------------------------------------------- > This is an automatically generated e-mail. To reply, visit: > https://reviews.apache.org/r/48213/ > ----------------------------------------------------------- > > (Updated June 3, 2016, 10:09 p.m.) > > > Review request for samza, Chris Pettitt, Navina Ramesh, and Yi Pan (Data > Infrastructure). > > > Repository: samza > > > Description > ------- > > All the system producers need to be thread safe in order to be used in > multithreaded tasks. The following are the changes > (ElasticSearchSystemProducer is already thread safe so no change made there): > > In KafkaSystemProducer, remove the buggy retry logic and treat any exception > as fatal. > In HdfsSystemProducer, add synchronization lock to all public methods. > > > Diffs > ----- > > > samza-hdfs/src/main/scala/org/apache/samza/system/hdfs/HdfsSystemProducer.scala > 1f4b5c46436e44b7c7cd1a49689c4f43f1f6ed1b > > samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemProducer.scala > 3769e103616dc0f1fd869706cc086e24cd926c48 > > samza-kafka/src/test/java/org/apache/samza/system/kafka/TestKafkaSystemProducerJava.java > 04c9113fd6c3dd56c49ff46c8c1c0ff12f68e5e2 > > samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestKafkaSystemProducer.scala > 8e32bba6ced090f0fc8d4e5176fe0788df36981d > > Diff: https://reviews.apache.org/r/48213/diff/ > > > Testing > ------- > > Unit tests and local testing. > > > Thanks, > > Xinyu Liu > >
