> On Aug. 19, 2014, 6:07 p.m., Chinmay Soman wrote: > > samza-core/src/main/scala/org/apache/samza/task/TaskInstanceCollector.scala, > > line 70 > > <https://reviews.apache.org/r/24820/diff/2/?file=664322#file664322line70> > > > > Do we need to track flushMs here ?
I opted not to track it here, but rather at the individual system implementation level (in the KafkaSystemProducer). The motivation for this is that flush is triggered not only when producerMultiplexer.flush is called. It's possible for the underlying system to flush at any point. In Kafka's case, we also flush when the buffer has reached some count threshold. If I tracked flushMs on the producerMultiplexer call, we'd lose out on timings when flushes happen due to a send() call. We *could* track both, but I opted against this since I thought the producerMultiplexer flush timings might be confusing, misleading, and redundant. - Chris ----------------------------------------------------------- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/24820/#review51000 ----------------------------------------------------------- On Aug. 19, 2014, 5:46 p.m., Chris Riccomini wrote: > > ----------------------------------------------------------- > This is an automatically generated e-mail. To reply, visit: > https://reviews.apache.org/r/24820/ > ----------------------------------------------------------- > > (Updated Aug. 19, 2014, 5:46 p.m.) > > > Review request for samza. > > > Bugs: SAMZA-384 > https://issues.apache.org/jira/browse/SAMZA-384 > > > Repository: samza > > > Description > ------- > > fix useless imports and wrong javadoc words > > > add javadocs. remove readable container. make all tests pass. > > > add task instance collector that sends immediately. > > > Diffs > ----- > > samza-core/src/main/scala/org/apache/samza/container/RunLoop.scala > 3a264ad4bd9ea6ffa801c662dae09ebd7ff79d32 > samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala > d574ac413c0ec81e12eb44b2d0cc0d9843aad434 > > samza-core/src/main/scala/org/apache/samza/container/SamzaContainerMetrics.scala > 44d5dffb36edd03032bbbd8c13541f18192f2ba2 > samza-core/src/main/scala/org/apache/samza/container/TaskInstance.scala > 9484ddb50a97eef52ab6ff7c932fdda0ff1ecb0a > > samza-core/src/main/scala/org/apache/samza/container/TaskInstanceMetrics.scala > aae3f8795ef9a12beaefa0917939107102e76b31 > samza-core/src/main/scala/org/apache/samza/task/ReadableCollector.scala > 444bf37db259d4fccc8ca2d479096c109911d46c > samza-core/src/main/scala/org/apache/samza/task/TaskInstanceCollector.scala > PRE-CREATION > samza-core/src/test/scala/org/apache/samza/container/TestRunLoop.scala > 86b7f31d1a574f4ee66e8498ecb501a08b94dff5 > > samza-core/src/test/scala/org/apache/samza/container/TestSamzaContainer.scala > e3c7fe3e2d329b0767eb439144b1ba419848bb96 > samza-core/src/test/scala/org/apache/samza/container/TestTaskInstance.scala > 9d5ff1309cd7ae6d41b48870ef7445d3710a2196 > > samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemProducer.scala > 22c8f0c2954d070aab9ce8d204aa8220d9c7bd74 > > samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemProducerMetrics.scala > ad39157c6d052b2e14e51b2f8a61d740fc18a129 > > samza-test/src/main/scala/org/apache/samza/test/performance/TestKeyValuePerformance.scala > 7d0b8db0c3bf1e70fd6af03abb594c7000e6666b > > samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaAppMasterMetrics.scala > 851aae6b347b0be2cd2d891fc45030c3e47189d4 > > Diff: https://reviews.apache.org/r/24820/diff/ > > > Testing > ------- > > > Thanks, > > Chris Riccomini > >
