public KsqlEngineMetrics(String metricGroupPrefix, KsqlEngine ksqlEngine) {
  this.ksqlEngine = ksqlEngine;
  this.sensors = new ArrayList<>();
  this.metricGroupName = metricGroupPrefix + "-query-stats";

  Metrics metrics = MetricCollectors.getMetrics();

  this.numActiveQueries = configureNumActiveQueries(metrics);
  this.messagesIn = configureMessagesIn(metrics);
  this.totalMessagesIn = configureTotalMessagesIn(metrics);
  this.totalBytesIn = configureTotalBytesIn(metrics);
  this.messagesOut =  configureMessagesOut(metrics);
  this.numIdleQueries = configureIdleQueriesSensor(metrics);
  this.messageConsumptionByQuery = configureMessageConsumptionByQuerySensor(metrics);
  this.errorRate = configureErrorRate(metrics);
}
 

public static void initialize() {
  MetricConfig metricConfig = new MetricConfig()
      .samples(100)
      .timeWindow(
          1000,
          TimeUnit.MILLISECONDS
      );
  List<MetricsReporter> reporters = new ArrayList<>();
  reporters.add(new JmxReporter("io.confluent.ksql.metrics"));
  // Replace all static contents other than Time to ensure they are cleaned for tests that are
  // not aware of the need to initialize/cleanup this test, in case test processes are reused.
  // Tests aware of the class clean everything up properly to get the state into a clean state,
  // a full, fresh instantiation here ensures something like KsqlEngineMetricsTest running after
  // another test that used MetricsCollector without running cleanUp will behave correctly.
  metrics = new Metrics(metricConfig, reporters, new SystemTime());
  collectorMap = new ConcurrentHashMap<>();
}
 

@Test
public void shouldDisplayRateThroughput() throws Exception {

  ConsumerCollector collector = new ConsumerCollector();//
  collector.configure(new Metrics(), "group", new SystemTime());

  for (int i = 0; i < 100; i++){

    Map<TopicPartition, List<ConsumerRecord<Object, Object>>> records = ImmutableMap.of(
            new TopicPartition(TEST_TOPIC, 1), Arrays.asList(
                    new ConsumerRecord<>(TEST_TOPIC, 1, i,  1l, TimestampType.CREATE_TIME,  1l, 10, 10, "key", "1234567890")) );
    ConsumerRecords<Object, Object> consumerRecords = new ConsumerRecords<>(records);

    collector.onConsume(consumerRecords);
  }

  Collection<TopicSensors.Stat> stats = collector.stats(TEST_TOPIC, false);
  assertNotNull(stats);

  assertThat( stats.toString(), containsString("name=consumer-messages-per-sec,"));
  assertThat( stats.toString(), containsString("total-messages, value=100.0"));
}
 

/**
 * Creates a new network client with the given properties.
 *
 * @return A new network client with the given properties.
 */
NetworkClient createNetworkClient(long connectionMaxIdleMS,
                                  Metrics metrics,
                                  Time time,
                                  String metricGrpPrefix,
                                  ChannelBuilder channelBuilder,
                                  Metadata metadata,
                                  String clientId,
                                  int maxInFlightRequestsPerConnection,
                                  long reconnectBackoffMs,
                                  long reconnectBackoffMax,
                                  int socketSendBuffer,
                                  int socketReceiveBuffer,
                                  int defaultRequestTimeoutMs,
                                  boolean discoverBrokerVersions,
                                  ApiVersions apiVersions);
 

@Override
public NetworkClient createNetworkClient(long connectionMaxIdleMS,
                                         Metrics metrics,
                                         Time time,
                                         String metricGrpPrefix,
                                         ChannelBuilder channelBuilder,
                                         Metadata metadata,
                                         String clientId,
                                         int maxInFlightRequestsPerConnection,
                                         long reconnectBackoffMs,
                                         long reconnectBackoffMax,
                                         int socketSendBuffer,
                                         int socketReceiveBuffer,
                                         int defaultRequestTimeoutMs,
                                         boolean discoverBrokerVersions,
                                         ApiVersions apiVersions) {
  return new NetworkClient(new Selector(connectionMaxIdleMS, metrics, time, metricGrpPrefix, channelBuilder, new LogContext()),
                           metadata, clientId, maxInFlightRequestsPerConnection, reconnectBackoffMs,
                           reconnectBackoffMax, socketSendBuffer, socketReceiveBuffer, defaultRequestTimeoutMs,
                           ClientDnsLookup.DEFAULT, time, discoverBrokerVersions, apiVersions, new LogContext());
}
 

public KafkaNodeClient(int id, String host, int port) {
	node = new Node(id, host, port);
	
	//
	LogContext logContext = new LogContext("ctx");

	ConfigDef defConf = new ConfigDef();
	defConf.define(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, ConfigDef.Type.STRING,
			CommonClientConfigs.DEFAULT_SECURITY_PROTOCOL, ConfigDef.Importance.MEDIUM,
			CommonClientConfigs.SECURITY_PROTOCOL_DOC);

	defConf.define(SaslConfigs.SASL_MECHANISM, ConfigDef.Type.STRING, SaslConfigs.DEFAULT_SASL_MECHANISM,
			ConfigDef.Importance.MEDIUM, SaslConfigs.SASL_MECHANISM_DOC);

	metrics = new Metrics(Time.SYSTEM);

	AbstractConfig config = new AbstractConfig(defConf, new Properties());
	channelBuilder = ClientUtils.createChannelBuilder(config);
	selector = new Selector(1000L, metrics, Time.SYSTEM, "cc", channelBuilder, logContext);
	client = new NetworkClient(selector, new Metadata(0, Long.MAX_VALUE, false),
			CLIENT_ID, 10, 1000L, 1000L, 1, 1024, 1000, Time.SYSTEM, true, new ApiVersions(),
			null, logContext);
}
 

/**
 * Metrics for Calculating the offset commit latency of a consumer.
 * @param metrics the commit offset metrics
 * @param tags the tags associated, i.e) kmf.services:name=single-cluster-monitor
 */
public CommitLatencyMetrics(Metrics metrics, Map<String, String> tags, int latencyPercentileMaxMs,
    int latencyPercentileGranularityMs) {
  _inProgressCommit = false;
  _commitOffsetLatency = metrics.sensor("commit-offset-latency");
  _commitOffsetLatency.add(new MetricName("commit-offset-latency-ms-avg", METRIC_GROUP_NAME, "The average latency in ms of committing offset", tags), new Avg());
  _commitOffsetLatency.add(new MetricName("commit-offset-latency-ms-max", METRIC_GROUP_NAME, "The maximum latency in ms of committing offset", tags), new Max());

  if (latencyPercentileGranularityMs == 0) {
    throw new IllegalArgumentException("The latency percentile granularity was incorrectly passed a zero value.");
  }

  // 2 extra buckets exist which are respectively designated for values which are less than 0.0 or larger than max.
  int bucketNum = latencyPercentileMaxMs / latencyPercentileGranularityMs + 2;
  int sizeInBytes = bucketNum * 4;
  _commitOffsetLatency.add(new Percentiles(sizeInBytes, latencyPercentileMaxMs, Percentiles.BucketSizing.CONSTANT,
      new Percentile(new MetricName("commit-offset-latency-ms-99th", METRIC_GROUP_NAME, "The 99th percentile latency of committing offset", tags), 99.0),
      new Percentile(new MetricName("commit-offset-latency-ms-999th", METRIC_GROUP_NAME, "The 99.9th percentile latency of committing offset", tags), 99.9),
      new Percentile(new MetricName("commit-offset-latency-ms-9999th", METRIC_GROUP_NAME, "The 99.99th percentile latency of committing offset", tags), 99.99)));
  LOG.info("{} was constructed successfully.", this.getClass().getSimpleName());
}
 

/**
 * Metrics for Calculating the offset commit availability of a consumer.
 * @param metrics the commit offset metrics
 * @param tags the tags associated, i.e) kmf.services:name=single-cluster-monitor
 */
public CommitAvailabilityMetrics(final Metrics metrics, final Map<String, String> tags) {
  LOG.info("{} called.", this.getClass().getSimpleName());
  _offsetsCommitted = metrics.sensor("offsets-committed");
  _offsetsCommitted.add(new MetricName("offsets-committed-total", METRIC_GROUP_NAME,
      "The total number of offsets per second that are committed.", tags), new Total());

  _failedCommitOffsets = metrics.sensor("failed-commit-offsets");
  _failedCommitOffsets.add(new MetricName("failed-commit-offsets-avg", METRIC_GROUP_NAME,
      "The average number of offsets per second that have failed.", tags), new Rate());
  _failedCommitOffsets.add(new MetricName("failed-commit-offsets-total", METRIC_GROUP_NAME,
      "The total number of offsets per second that have failed.", tags), new Total());

  metrics.addMetric(new MetricName("offsets-committed-avg", METRIC_GROUP_NAME, "The average offset commits availability.", tags),
    (MetricConfig config, long now) -> {
      Object offsetCommitTotal = metrics.metrics().get(metrics.metricName("offsets-committed-total", METRIC_GROUP_NAME, tags)).metricValue();
      Object offsetCommitFailTotal = metrics.metrics().get(metrics.metricName("failed-commit-offsets-total", METRIC_GROUP_NAME, tags)).metricValue();
      if (offsetCommitTotal != null && offsetCommitFailTotal != null) {
        double offsetsCommittedCount = (double) offsetCommitTotal;
        double offsetsCommittedErrorCount = (double) offsetCommitFailTotal;
        return offsetsCommittedCount / (offsetsCommittedCount + offsetsCommittedErrorCount);
      } else {
        return 0;
      }
    });
}
 

/**
 *
 * @param metrics a named, numerical measurement. sensor is a handle to record numerical measurements as they occur.
 * @param tags metrics/sensor's tags
 */
public ClusterTopicManipulationMetrics(final Metrics metrics, final Map<String, String> tags) {
  super(metrics, tags);
  _topicCreationSensor = metrics.sensor("topic-creation-metadata-propagation");
  _topicDeletionSensor = metrics.sensor("topic-deletion-metadata-propagation");
  _topicCreationSensor.add(new MetricName("topic-creation-metadata-propagation-ms-avg", METRIC_GROUP_NAME,
      "The average propagation duration in ms of propagating topic creation data and metadata to all brokers in the cluster",
      tags), new Avg());
  _topicCreationSensor.add(new MetricName("topic-creation-metadata-propagation-ms-max", METRIC_GROUP_NAME,
      "The maximum propagation time in ms of propagating topic creation data and metadata to all brokers in the cluster",
      tags), new Max());
  _topicDeletionSensor.add(new MetricName("topic-deletion-metadata-propagation-ms-avg", METRIC_GROUP_NAME,
      "The average propagation duration in milliseconds of propagating the topic deletion data and metadata "
          + "across all the brokers in the cluster.", tags), new Avg());
  _topicDeletionSensor.add(new MetricName("topic-deletion-metadata-propagation-ms-max", METRIC_GROUP_NAME,
      "The maximum propagation time in milliseconds of propagating the topic deletion data and metadata "
          + "across all the brokers in the cluster.", tags), new Max());

  LOGGER.debug("{} constructor was initialized successfully.", "ClusterTopicManipulationMetrics");
}
 

public ClusterTopicManipulationService(String name, AdminClient adminClient) {
  LOGGER.info("ClusterTopicManipulationService constructor initiated {}", this.getClass().getName());

  _isOngoingTopicCreationDone = true;
  _isOngoingTopicDeletionDone = true;
  _adminClient = adminClient;
  _executor = Executors.newSingleThreadScheduledExecutor();
  _reportIntervalSecond = Duration.ofSeconds(1);
  _running = new AtomicBoolean(false);
  _configDefinedServiceName = name;
  // TODO: instantiate a new instance of ClusterTopicManipulationMetrics(..) here.

  MetricConfig metricConfig = new MetricConfig().samples(60).timeWindow(1000, TimeUnit.MILLISECONDS);
  List<MetricsReporter> reporters = new ArrayList<>();
  reporters.add(new JmxReporter(Service.JMX_PREFIX));
  Metrics metrics = new Metrics(metricConfig, reporters, new SystemTime());

  Map<String, String> tags = new HashMap<>();
  tags.put("name", name);
  _clusterTopicManipulationMetrics = new ClusterTopicManipulationMetrics(metrics, tags);
}
 

@Test
public void commitAvailabilityTest() throws Exception {
  ConsumeService consumeService = consumeService();
  Metrics metrics = consumeServiceMetrics(consumeService);

  Assert.assertNotNull(metrics.metrics().get(metrics.metricName("offsets-committed-total", METRIC_GROUP_NAME, tags)).metricValue());
  Assert.assertEquals(metrics.metrics().get(metrics.metricName("offsets-committed-total", METRIC_GROUP_NAME, tags)).metricValue(), 0.0);

  /* Should start */
  consumeService.startConsumeThreadForTesting();
  Assert.assertTrue(consumeService.isRunning());

  /* in milliseconds */
  long threadStartDelay = TimeUnit.SECONDS.toMillis(THREAD_START_DELAY_SECONDS);

  /* Thread.sleep safe to do here instead of ScheduledExecutorService
  *  We want to sleep current thread so that consumeService can start running for enough seconds. */
  Thread.sleep(threadStartDelay);
  Assert.assertNotNull(metrics.metrics().get(metrics.metricName("offsets-committed-total", METRIC_GROUP_NAME, tags)).metricValue());
  Assert.assertNotNull(metrics.metrics().get(metrics.metricName("failed-commit-offsets-total", METRIC_GROUP_NAME,
      tags)).metricValue());
  Assert.assertEquals(metrics.metrics().get(metrics.metricName("failed-commit-offsets-total", METRIC_GROUP_NAME, tags)).metricValue(), 0.0);
  Assert.assertNotEquals(metrics.metrics().get(metrics.metricName("offsets-committed-total", METRIC_GROUP_NAME, tags)).metricValue(), 0.0);
  shutdownConsumeService(consumeService);
}
 

@Test
public void commitLatencyTest() throws Exception {
  CommitLatencyMetrics commitLatencyMetrics = Mockito.mock(CommitLatencyMetrics.class);
  Assert.assertNotNull(commitLatencyMetrics);

  ConsumeService consumeService = consumeService();
  Metrics metrics = consumeServiceMetrics(consumeService);

  Assert.assertNull(metrics.metrics().get(metrics.metricName("commit-offset-latency-ms-avg", METRIC_GROUP_NAME, tags)));
  Assert.assertNull(metrics.metrics().get(metrics.metricName("commit-offset-latency-ms-max", METRIC_GROUP_NAME, tags)));

  /* Should start */
  consumeService.startConsumeThreadForTesting();
  Assert.assertTrue(consumeService.isRunning());

  /* in milliseconds */
  long threadStartDelay = TimeUnit.SECONDS.toMillis(THREAD_START_DELAY_SECONDS);

  /* Thread.sleep safe to do here instead of ScheduledExecutorService
   *  We want to sleep current thread so that consumeService can start running for enough seconds. */
  Thread.sleep(threadStartDelay);

  shutdownConsumeService(consumeService);
}
 

@Test
public void testMetricChange() throws Exception {
    Metrics metrics = new Metrics();
    DropwizardReporter reporter = new DropwizardReporter();
    reporter.configure(new HashMap<String, Object>());
    metrics.addReporter(reporter);
    Sensor sensor = metrics.sensor("kafka.requests");
    sensor.add(new MetricName("pack.bean1.avg", "grp1"), new Avg());

    Map<String, Gauge> gauges = SharedMetricRegistries.getOrCreate("default").getGauges();
    String expectedName = "org.apache.kafka.common.metrics.grp1.pack.bean1.avg";
    Assert.assertEquals(1, gauges.size());
    Assert.assertEquals(expectedName, gauges.keySet().toArray()[0]);

    sensor.record(2.1);
    sensor.record(2.2);
    sensor.record(2.6);
    Assert.assertEquals(2.3, (Double)gauges.get(expectedName).getValue(), 0.001);
}
 

/**
 * Initialize the coordination manager.
 */
public KarelDbCoordinator(
    LogContext logContext,
    ConsumerNetworkClient client,
    String groupId,
    int rebalanceTimeoutMs,
    int sessionTimeoutMs,
    int heartbeatIntervalMs,
    Metrics metrics,
    String metricGrpPrefix,
    Time time,
    long retryBackoffMs,
    KarelDbIdentity identity,
    KarelDbRebalanceListener listener) {
    super(
        new GroupRebalanceConfig(
            sessionTimeoutMs,
            rebalanceTimeoutMs,
            heartbeatIntervalMs,
            groupId,
            Optional.empty(),
            retryBackoffMs,
            true
        ),
        logContext,
        client,
        metrics,
        metricGrpPrefix,
        time
    );
    this.identity = identity;
    this.assignmentSnapshot = null;
    this.listener = listener;
}
 

private Sensor configureErrorRate(Metrics metrics) {
  Sensor sensor = createSensor(metrics, metricGroupName + "-error-rate");
  sensor.add(
      metrics.metricName("error-rate", this.metricGroupName,
                         "The number of messages which were consumed but not processed. "
                         + "Messages may not be processed if, for instance, the message "
                         + "contents could not be deserialized due to an incompatible schema. "
                         + "Alternately, a consumed messages may not have been produced, hence "
                         + "being effectively dropped. Such messages would also be counted "
                         + "toward the error rate."),
      new Value());
  return sensor;
}
 

private Sensor configureMessagesOut(Metrics metrics) {
  Sensor sensor = createSensor(metrics, metricGroupName + "-messages-produced");
  sensor.add(
      metrics.metricName("messages-produced-per-sec", this.metricGroupName,
                         "The number of messages produced per second across all queries"),
      new Value());

  return sensor;
}
 

private Sensor configureMessagesIn(Metrics metrics) {
  Sensor sensor = createSensor(metrics, metricGroupName + "-messages-consumed");
  sensor.add(
      metrics.metricName("messages-consumed-per-sec", this.metricGroupName,
                         "The number of messages consumed per second across all queries"),
      new Value());
  return sensor;
}
 

private Sensor configureTotalMessagesIn(Metrics metrics) {
  Sensor sensor = createSensor(metrics, metricGroupName + "-total-messages-consumed");
  sensor.add(
      metrics.metricName("messages-consumed-total", this.metricGroupName,
          "The total number of messages consumed across all queries"),
      new Value());
  return sensor;
}
 

private Sensor configureTotalBytesIn(Metrics metrics) {
  Sensor sensor = createSensor(metrics, metricGroupName + "-total-bytes-consumed");
  sensor.add(
      metrics.metricName("bytes-consumed-total", this.metricGroupName,
          "The total number of bytes consumed across all queries"),
      new Value());
  return sensor;
}
 

private Sensor configureMessageConsumptionByQuerySensor(Metrics metrics) {
  Sensor sensor = createSensor(metrics, "message-consumption-by-query");
  sensor.add(metrics.metricName("messages-consumed-max", this.metricGroupName), new Max());
  sensor.add(metrics.metricName("messages-consumed-min", this.metricGroupName), new Min());
  sensor.add(metrics.metricName("messages-consumed-avg", this.metricGroupName), new Avg());
  return sensor;
}
 

@Test
public void shouldRemoveAllSensorsOnClose() {
  assertTrue(engineMetrics.registeredSensors().size() > 0);

  engineMetrics.close();

  Metrics metrics = MetricCollectors.getMetrics();
  engineMetrics.registeredSensors().forEach(sensor -> {
    assertTrue(metrics.getSensor(sensor.name()) == null);
  });
}
 

@Test
public void shouldRecordNumberOfActiveQueries() {
  EasyMock.expect(ksqlEngine.numberOfLiveQueries()).andReturn(3L);
  EasyMock.replay(ksqlEngine);
  Metrics metrics = MetricCollectors.getMetrics();
  double value = getMetricValue(metrics, "num-active-queries");
  assertEquals(3.0, value, 0.0);
}
 

@Test
public void shouldRecordNumberOfPersistentQueries() {
  EasyMock.expect(ksqlEngine.numberOfPersistentQueries()).andReturn(3L);
  EasyMock.replay(ksqlEngine);
  Metrics metrics = MetricCollectors.getMetrics();
  double value = getMetricValue(metrics, "num-persistent-queries");
  assertEquals(3.0, value, 0.0);
}
 

@Test
public void shouldRecordMessagesConsumed() {
  int numMessagesConsumed = 500;
  consumeMessages(numMessagesConsumed, "group1");
  Metrics metrics = MetricCollectors.getMetrics();
  engineMetrics.updateMetrics();
  double value = getMetricValue(metrics, "messages-consumed-per-sec");
  assertEquals(numMessagesConsumed / 100, Math.floor(value), 0.01);
}
 

@Test
public void shouldRecordMessagesProduced() {
  int numMessagesProduced = 500;
  produceMessages(numMessagesProduced);
  Metrics metrics = MetricCollectors.getMetrics();
  engineMetrics.updateMetrics();
  double value = getMetricValue(metrics, "messages-produced-per-sec");
  assertEquals(numMessagesProduced / 100, Math.floor(value), 0.01);
}
 

@Test
public void shouldRecordMessagesConsumedByQuery() {
  int numMessagesConsumed = 500;
  consumeMessages(numMessagesConsumed, "group1");
  consumeMessages(numMessagesConsumed * 100, "group2");
  Metrics metrics = MetricCollectors.getMetrics();
  engineMetrics.updateMetrics();
  double maxValue = getMetricValue(metrics, "messages-consumed-max");
  assertEquals(numMessagesConsumed, Math.floor(maxValue), 5.0);
  double minValue = getMetricValue(metrics, "messages-consumed-min");
  assertEquals(numMessagesConsumed / 100, Math.floor(minValue), 0.01);
}
 

@Test
public void shouldRecordErrors() throws Exception {

  ProducerCollector collector = new ProducerCollector().configure(new Metrics(), "clientid", MetricCollectors.getTime());

  for (int i = 0; i < 1000; i++){
    collector.recordError(TEST_TOPIC);
  }

  Collection<TopicSensors.Stat> stats = collector.stats("test-topic", true);

  assertThat( stats.toString(), containsString("failed-messages"));
  assertThat( stats.toString(), containsString("value=1000"));
}
 

/**
 * Initialize the coordination manager.
 */
public WorkerCoordinator(ConsumerNetworkClient client,
                         String groupId,
                         int rebalanceTimeoutMs,
                         int sessionTimeoutMs,
                         int heartbeatIntervalMs,
                         Metrics metrics,
                         String metricGrpPrefix,
                         Time time,
                         long retryBackoffMs,
                         String restUrl,
                         TaskConfigManager taskConfigManager,
                         WorkerRebalanceListener listener) {
    super(client,
            groupId,
            rebalanceTimeoutMs,
            sessionTimeoutMs,
            heartbeatIntervalMs,
            metrics,
            metricGrpPrefix,
            time,
            retryBackoffMs);
    this.restUrl = restUrl;
    this.taskConfigManager = taskConfigManager;
    this.assignmentSnapshot = null;
    this.sensors = new WorkerCoordinatorMetrics(metrics, metricGrpPrefix);
    this.listener = listener;
    this.rejoinRequested = false;
}
 

public WorkerCoordinatorMetrics(Metrics metrics, String metricGrpPrefix) {
    this.metrics = metrics;
    this.metricGrpName = metricGrpPrefix + "-coordinator-metrics";

    Measurable numTasks = (config, now) -> assignmentSnapshot.tasks().size();
    metrics.addMetric(metrics.metricName("assigned-tasks",
            this.metricGrpName,
            "The number of tasks currently assigned to this consumer"), numTasks);
}
 

ConnectorJmxReporter(Metrics metrics) {
  super(metrics);
  allStates.put("RUNNING", "running");
  allStates.put("FAILED", "failed");
  allStates.put("DESTROYED", "destroyed");
  allStates.put("UNASSIGNED", "unassigned");
  allStates.put("PAUSED", "paused");
  connectorLevelJmxTags.add(CONNECTOR_KEY);
}