abhishekshivanna commented on a change in pull request #1368:
URL: https://github.com/apache/samza/pull/1368#discussion_r432628091
##########
File path:
samza-core/src/main/scala/org/apache/samza/metrics/reporter/MetricsSnapshotReporterFactory.scala
##########
@@ -19,53 +19,47 @@
package org.apache.samza.metrics.reporter
-import org.apache.samza.util.{Logging, ReflectionUtil, StreamUtil, Util}
import org.apache.samza.SamzaException
-import org.apache.samza.config.{Config, JobConfig, MetricsConfig,
SerializerConfig, StreamConfig, SystemConfig}
-import org.apache.samza.metrics.MetricsReporter
-import org.apache.samza.metrics.MetricsReporterFactory
-import org.apache.samza.metrics.MetricsRegistryMap
-import org.apache.samza.serializers.{MetricsSnapshotSerdeV2, SerdeFactory}
-import org.apache.samza.system.SystemFactory
+import org.apache.samza.config._
+import org.apache.samza.metrics.{MetricsRegistryMap, MetricsReporter,
MetricsReporterFactory}
+import org.apache.samza.serializers.{MetricsSnapshotSerdeV2, Serde,
SerdeFactory}
+import org.apache.samza.system.{SystemFactory, SystemProducer, SystemStream}
import org.apache.samza.util.ScalaJavaUtil.JavaOptionals
+import org.apache.samza.util.{Logging, ReflectionUtil, StreamUtil, Util}
class MetricsSnapshotReporterFactory extends MetricsReporterFactory with
Logging {
- def getMetricsReporter(name: String, containerName: String, config: Config):
MetricsReporter = {
- info("Creating new metrics snapshot reporter.")
-
- val jobConfig = new JobConfig(config)
- val jobName = JavaOptionals.toRichOptional(jobConfig.getName).toOption
- .getOrElse(throw new SamzaException("Job name must be defined in
config."))
- val jobId = jobConfig.getJobId
-
- val metricsConfig = new MetricsConfig(config)
- val metricsSystemStreamName =
JavaOptionals.toRichOptional(metricsConfig.getMetricsSnapshotReporterStream(name))
- .toOption
- .getOrElse(throw new SamzaException("No metrics stream defined in
config."))
-
- val systemStream =
StreamUtil.getSystemStreamFromNames(metricsSystemStreamName)
-
- info("Got system stream %s." format systemStream)
-
- val systemName = systemStream.getSystem
+ def getProducer(reporterName: String, config: Config, registry:
MetricsRegistryMap): SystemProducer = {
Review comment:
This kinda relates to @bharathkk comment - where we need to share this
producer externally in `DiagnosticsUtil`. See changes in
https://github.com/apache/samza/pull/1369/files
Do you think if we had two separate producers for `DiagnosticsManager` and
`MetricsSnapshotReporter` would pose a problem?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
