@Test public void testPartitionManagedLockAcquireFailMetric() throws InterruptedException { String datastreamName = "pizzaStream"; Datastream datastream = KafkaMirrorMakerConnectorTestUtils.createDatastream(datastreamName, _broker, "\\w+Pizza"); DatastreamTaskImpl task = spy(new DatastreamTaskImpl(Collections.singletonList(datastream))); doThrow(DatastreamRuntimeException.class).when(task).acquire(any(Duration.class)); MockDatastreamEventProducer datastreamProducer = new MockDatastreamEventProducer(); task.setEventProducer(datastreamProducer); KafkaBasedConnectorConfig connectorConfig = new KafkaBasedConnectorConfigBuilder() .setConsumerFactory(new LiKafkaConsumerFactory()) .setCommitIntervalMillis(10000) .setEnablePartitionManaged(true) .build(); createAndConnectZkAdapter(task); String connectorName = "KafkaMirrorMaker"; KafkaMirrorMakerConnectorTaskTest connectorTask = new KafkaMirrorMakerConnectorTaskTest(connectorConfig, task, connectorName, false, new KafkaMirrorMakerGroupIdConstructor(false, "testCluster")); AtomicReference<Throwable> throwable = new AtomicReference<>(); Thread connectorThread = KafkaMirrorMakerConnectorTestUtils.runKafkaMirrorMakerConnectorTask(connectorTask, (t1, e) -> throwable.set(e), false); connectorThread.join(); Assert.assertEquals(DatastreamRuntimeException.class, throwable.get().getClass()); Meter metric = DynamicMetricsManager.getInstance() .getMetric(connectorName + "." + KafkaMirrorMakerConnectorTask.class.getSimpleName() + "." + datastreamName + "." + "taskLockAcquireErrorRate"); Assert.assertNotNull(metric); Assert.assertEquals(metric.getCount(), 1); MetricsTestUtils.verifyMetrics(new MetricsAware() { @Override public List<BrooklinMetricInfo> getMetricInfos() { return KafkaMirrorMakerConnectorTaskTest.getMetricInfos(connectorName); } }, DynamicMetricsManager.getInstance()); }
public static List<BrooklinMetricInfo> getMetricInfos(String connectorName) { List<BrooklinMetricInfo> metrics = new ArrayList<>(); metrics.addAll(AbstractKafkaBasedConnectorTask.getMetricInfos( generateMetricsPrefix(connectorName, CLASS_NAME) + MetricsAware.KEY_REGEX)); metrics.add(new BrooklinMeterInfo(generateMetricsPrefix(connectorName, CLASS_NAME) + MetricsAware.KEY_REGEX + TASK_LOCK_ACQUIRE_ERROR_RATE)); metrics.add(new BrooklinCounterInfo( generateMetricsPrefix(connectorName, CLASS_NAME) + TOPIC_MANAGER_METRICS_PREFIX + MetricsAware.KEY_REGEX + ".*")); metrics.add(new BrooklinGaugeInfo( generateMetricsPrefix(connectorName, CLASS_NAME) + TOPIC_MANAGER_METRICS_PREFIX + MetricsAware.KEY_REGEX + ".*")); return metrics; }
KafkaMirrorMakerConnectorTask extends AbstractKafkaBasedConnectorTask { public static List<BrooklinMetricInfo> getMetricInfos(String connectorName) { List<BrooklinMetricInfo> metrics = new ArrayList<>(); metrics.addAll(AbstractKafkaBasedConnectorTask.getMetricInfos( generateMetricsPrefix(connectorName, CLASS_NAME) + MetricsAware.KEY_REGEX)); metrics.add(new BrooklinMeterInfo(generateMetricsPrefix(connectorName, CLASS_NAME) + MetricsAware.KEY_REGEX + TASK_LOCK_ACQUIRE_ERROR_RATE)); metrics.add(new BrooklinCounterInfo( generateMetricsPrefix(connectorName, CLASS_NAME) + TOPIC_MANAGER_METRICS_PREFIX + MetricsAware.KEY_REGEX + ".*")); metrics.add(new BrooklinGaugeInfo( generateMetricsPrefix(connectorName, CLASS_NAME) + TOPIC_MANAGER_METRICS_PREFIX + MetricsAware.KEY_REGEX + ".*")); return metrics; } }
KafkaMirrorMakerConnectorTask extends AbstractKafkaBasedConnectorTask { public static List<BrooklinMetricInfo> getMetricInfos(String connectorName) { List<BrooklinMetricInfo> metrics = new ArrayList<>(); metrics.addAll(AbstractKafkaBasedConnectorTask.getMetricInfos( generateMetricsPrefix(connectorName, CLASS_NAME) + MetricsAware.KEY_REGEX)); metrics.add(new BrooklinMeterInfo(generateMetricsPrefix(connectorName, CLASS_NAME) + MetricsAware.KEY_REGEX + TASK_LOCK_ACQUIRE_ERROR_RATE)); metrics.add(new BrooklinCounterInfo( generateMetricsPrefix(connectorName, CLASS_NAME) + TOPIC_MANAGER_METRICS_PREFIX + MetricsAware.KEY_REGEX + ".*")); metrics.add(new BrooklinGaugeInfo( generateMetricsPrefix(connectorName, CLASS_NAME) + TOPIC_MANAGER_METRICS_PREFIX + MetricsAware.KEY_REGEX + ".*")); return metrics; } KafkaMirrorMakerConnectorTask(KafkaBasedConnectorConfig config, DatastreamTask task, String connectorName,
      boolean isFlushlessModeEnabled, GroupIdConstructor groupIdConstructor); }
KafkaMirrorMakerConnectorTask extends AbstractKafkaBasedConnectorTask { public static List<BrooklinMetricInfo> getMetricInfos(String connectorName) { List<BrooklinMetricInfo> metrics = new ArrayList<>(); metrics.addAll(AbstractKafkaBasedConnectorTask.getMetricInfos( generateMetricsPrefix(connectorName, CLASS_NAME) + MetricsAware.KEY_REGEX)); metrics.add(new BrooklinMeterInfo(generateMetricsPrefix(connectorName, CLASS_NAME) + MetricsAware.KEY_REGEX + TASK_LOCK_ACQUIRE_ERROR_RATE)); metrics.add(new BrooklinCounterInfo( generateMetricsPrefix(connectorName, CLASS_NAME) + TOPIC_MANAGER_METRICS_PREFIX + MetricsAware.KEY_REGEX + ".*")); metrics.add(new BrooklinGaugeInfo( generateMetricsPrefix(connectorName, CLASS_NAME) + TOPIC_MANAGER_METRICS_PREFIX + MetricsAware.KEY_REGEX + ".*")); return metrics; } KafkaMirrorMakerConnectorTask(KafkaBasedConnectorConfig config, DatastreamTask task, String connectorName,
      boolean isFlushlessModeEnabled, GroupIdConstructor groupIdConstructor); @Override void onPartitionsAssigned(Collection<TopicPartition> partitions); @Override void run(); @Override void stop(); @Override void onPartitionsRevoked(Collection<TopicPartition> partitions); static List<BrooklinMetricInfo> getMetricInfos(String connectorName); @Override KafkaDatastreamStatesResponse getKafkaDatastreamStatesResponse(); @VisibleForTesting TopicManager getTopicManager(); }
KafkaMirrorMakerConnectorTask extends AbstractKafkaBasedConnectorTask { public static List<BrooklinMetricInfo> getMetricInfos(String connectorName) { List<BrooklinMetricInfo> metrics = new ArrayList<>(); metrics.addAll(AbstractKafkaBasedConnectorTask.getMetricInfos( generateMetricsPrefix(connectorName, CLASS_NAME) + MetricsAware.KEY_REGEX)); metrics.add(new BrooklinMeterInfo(generateMetricsPrefix(connectorName, CLASS_NAME) + MetricsAware.KEY_REGEX + TASK_LOCK_ACQUIRE_ERROR_RATE)); metrics.add(new BrooklinCounterInfo( generateMetricsPrefix(connectorName, CLASS_NAME) + TOPIC_MANAGER_METRICS_PREFIX + MetricsAware.KEY_REGEX + ".*")); metrics.add(new BrooklinGaugeInfo( generateMetricsPrefix(connectorName, CLASS_NAME) + TOPIC_MANAGER_METRICS_PREFIX + MetricsAware.KEY_REGEX + ".*")); return metrics; } KafkaMirrorMakerConnectorTask(KafkaBasedConnectorConfig config, DatastreamTask task, String connectorName,
      boolean isFlushlessModeEnabled, GroupIdConstructor groupIdConstructor); @Override void onPartitionsAssigned(Collection<TopicPartition> partitions); @Override void run(); @Override void stop(); @Override void onPartitionsRevoked(Collection<TopicPartition> partitions); static List<BrooklinMetricInfo> getMetricInfos(String connectorName); @Override KafkaDatastreamStatesResponse getKafkaDatastreamStatesResponse(); @VisibleForTesting TopicManager getTopicManager(); static final String TOPIC_MANAGER_FACTORY; static final String DEFAULT_TOPIC_MANAGER_FACTORY; static final String DESTINATION_TOPIC_PREFIX; static final String DOMAIN_TOPIC_MANAGER; static final String TOPIC_MANAGER_METRICS_PREFIX; }