@Test @SuppressWarnings("unchecked") public void testAutoPauseOnSendFailure() throws Exception { String yummyTopic = "YummyPizza"; createTopic(_zkUtils, yummyTopic); Datastream datastream = KafkaMirrorMakerConnectorTestUtils.createDatastream("pizzaStream", _broker, "\\w+Pizza"); DatastreamTaskImpl task = new DatastreamTaskImpl(Collections.singletonList(datastream)); MockDatastreamEventProducer datastreamProducer = new MockDatastreamEventProducer((r) -> new String((byte[]) r.getEvents().get(0).key().get()).equals("key-2")); task.setEventProducer(datastreamProducer); Properties consumerProps = KafkaMirrorMakerConnectorTestUtils.getKafkaConsumerProperties(); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); KafkaMirrorMakerConnectorTask connectorTask = KafkaMirrorMakerConnectorTestUtils.createKafkaMirrorMakerConnectorTask(task, KafkaMirrorMakerConnectorTestUtils.getKafkaBasedConnectorConfigBuilder().setConsumerProps(consumerProps).build()); KafkaMirrorMakerConnectorTestUtils.runKafkaMirrorMakerConnectorTask(connectorTask); KafkaMirrorMakerConnectorTestUtils.produceEvents(yummyTopic, 5, _kafkaCluster); String autoPausedOnErrorMetricName = MetricRegistry.name(KafkaMirrorMakerConnectorTask.class.getSimpleName(), datastream.getName(), NUM_AUTO_PAUSED_PARTITIONS_ON_ERROR); if (!PollUtils.poll(() -> connectorTask.getAutoPausedSourcePartitions().contains(new TopicPartition(yummyTopic, 0)) && ((Gauge<Long>) DynamicMetricsManager.getInstance().getMetric(autoPausedOnErrorMetricName)).getValue() == 1, POLL_PERIOD_MS, POLL_TIMEOUT_MS)) { Assert.fail("Partition did not auto-pause after multiple send failures."); } Assert.assertEquals(datastreamProducer.getEvents().size(), 2, "The events before the failure should have been sent"); Map<String, Set<String>> pausedPartitions = new HashMap<>(); pausedPartitions.put(yummyTopic, new HashSet<>(Collections.singletonList("0"))); datastream.getMetadata() .put(DatastreamMetadataConstants.PAUSED_SOURCE_PARTITIONS_KEY, JsonUtils.toJson(pausedPartitions)); task = new DatastreamTaskImpl(Collections.singletonList(datastream)); connectorTask.checkForUpdateTask(task); String manualPausedMetricName = MetricRegistry.name(KafkaMirrorMakerConnectorTask.class.getSimpleName(), datastream.getName(), NUM_CONFIG_PAUSED_PARTITIONS); if (!PollUtils.poll(() -> pausedPartitions.equals(connectorTask.getPausedPartitionsConfig()) && ((Gauge<Long>) DynamicMetricsManager.getInstance().getMetric(manualPausedMetricName)).getValue() == 1 && connectorTask.getAutoPausedSourcePartitions().isEmpty() && ((Gauge<Long>) DynamicMetricsManager.getInstance().getMetric(autoPausedOnErrorMetricName)).getValue() == 0, POLL_PERIOD_MS, POLL_TIMEOUT_MS)) { Assert.fail("Paused partitions were not updated after adding partitions to pause config."); } datastreamProducer.setSendFailCondition((r) -> false); pausedPartitions.put(yummyTopic, Collections.emptySet()); datastream.getMetadata() .put(DatastreamMetadataConstants.PAUSED_SOURCE_PARTITIONS_KEY, JsonUtils.toJson(pausedPartitions)); task = new DatastreamTaskImpl(Collections.singletonList(datastream)); connectorTask.checkForUpdateTask(task); if (!PollUtils.poll(() -> pausedPartitions.equals(connectorTask.getPausedPartitionsConfig()) && ((Gauge<Long>) DynamicMetricsManager.getInstance().getMetric(manualPausedMetricName)).getValue() == 0, POLL_PERIOD_MS, POLL_TIMEOUT_MS)) { Assert.fail("Paused partitions were not updated after removing partitions from pause config."); } if (!PollUtils.poll(() -> datastreamProducer.getEvents().size() == 7, POLL_PERIOD_MS, POLL_TIMEOUT_MS)) { Assert.fail("datastream producer " + datastreamProducer.getEvents().size()); } connectorTask.stop(); Assert.assertTrue(connectorTask.awaitStop(CONNECTOR_AWAIT_STOP_TIMEOUT_MS, TimeUnit.MILLISECONDS), "did not shut down on time"); }
@Override public void stop() { super.stop(); _topicManager.stop(); }
KafkaMirrorMakerConnectorTask extends AbstractKafkaBasedConnectorTask { @Override public void stop() { super.stop(); _topicManager.stop(); } }
KafkaMirrorMakerConnectorTask extends AbstractKafkaBasedConnectorTask { @Override public void stop() { super.stop(); _topicManager.stop(); } KafkaMirrorMakerConnectorTask(KafkaBasedConnectorConfig config, DatastreamTask task, String connectorName,
      boolean isFlushlessModeEnabled, GroupIdConstructor groupIdConstructor); }
KafkaMirrorMakerConnectorTask extends AbstractKafkaBasedConnectorTask { @Override public void stop() { super.stop(); _topicManager.stop(); } 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 { @Override public void stop() { super.stop(); _topicManager.stop(); } 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; }