@Test public void testAutoPauseOnSendError() throws Exception { String[] cookieTopics = new String[] { "MacadamiaWhiteChocolateCookie", "AlmondCookie", "ChocolateChipCookie", "SaltyCaramelCookie" }; for (String topic : cookieTopics) { createTopic(_zkUtils, topic); } Datastream datastream = KafkaMirrorMakerConnectorTestUtils.createDatastream("cookieStream", _broker, "\\w+Cookie"); DatastreamTaskImpl task = new DatastreamTaskImpl(Collections.singletonList(datastream)); MockDatastreamEventProducer datastreamProducer = new MockDatastreamEventProducer((r) -> true); task.setEventProducer(datastreamProducer); KafkaMirrorMakerConnectorTask connectorTask = KafkaMirrorMakerConnectorTestUtils.createFlushlessKafkaMirrorMakerConnectorTask(task, true, 50, 100, Duration.ofDays(1)); KafkaMirrorMakerConnectorTestUtils.runKafkaMirrorMakerConnectorTask(connectorTask); for (String topic : cookieTopics) { KafkaMirrorMakerConnectorTestUtils.produceEvents(topic, 1, _kafkaCluster); } Assert.assertTrue(PollUtils.poll(() -> connectorTask.getAutoPausedSourcePartitions().size() == cookieTopics.length, POLL_PERIOD_MS, POLL_TIMEOUT_MS), "All topics should have had auto-paused partitions"); Assert.assertTrue(datastreamProducer.getEvents().isEmpty(), "No events should have been successfully sent"); 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; }