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; }
@Test public void testConsumeFromMultipleTopicsWithDestinationTopicPrefixMetadata() throws Exception { String yummyTopic = "YummyPizza"; String saltyTopic = "SaltyPizza"; String saladTopic = "HealthySalad"; String destinationTopicPrefixOverride = "newPrefix"; createTopic(_zkUtils, saladTopic); createTopic(_zkUtils, yummyTopic); createTopic(_zkUtils, saltyTopic); Datastream datastream = KafkaMirrorMakerConnectorTestUtils.createDatastream("pizzaStream", _broker, "\\w+Pizza"); datastream.getMetadata().put(DatastreamMetadataConstants.DESTINATION_TOPIC_PREFIX, destinationTopicPrefixOverride); DatastreamTaskImpl task = new DatastreamTaskImpl(Collections.singletonList(datastream)); MockDatastreamEventProducer datastreamProducer = new MockDatastreamEventProducer(); task.setEventProducer(datastreamProducer); KafkaMirrorMakerConnectorTask connectorTask = KafkaMirrorMakerConnectorTestUtils.createKafkaMirrorMakerConnectorTask(task); KafkaMirrorMakerConnectorTestUtils.runKafkaMirrorMakerConnectorTask(connectorTask); KafkaMirrorMakerConnectorTestUtils.produceEvents(yummyTopic, 1, _kafkaCluster); KafkaMirrorMakerConnectorTestUtils.produceEvents(saltyTopic, 1, _kafkaCluster); KafkaMirrorMakerConnectorTestUtils.produceEvents(saladTopic, 1, _kafkaCluster); if (!PollUtils.poll(() -> datastreamProducer.getEvents().size() == 2, POLL_PERIOD_MS, POLL_TIMEOUT_MS)) { Assert.fail("did not transfer the msgs within timeout. transferred " + datastreamProducer.getEvents().size()); } List<DatastreamProducerRecord> records = datastreamProducer.getEvents(); for (DatastreamProducerRecord record : records) { String destinationTopic = record.getDestination().get(); Assert.assertTrue(destinationTopic.endsWith("Pizza"), "Unexpected event consumed from Datastream and sent to topic: " + destinationTopic); Assert.assertTrue(destinationTopic.startsWith(destinationTopicPrefixOverride), "Destination topic prefix enabled, topic should start with prefix. Topic: " + destinationTopic); } validateTaskConsumerAssignment(connectorTask, Sets.newHashSet(new TopicPartition(yummyTopic, 0), new TopicPartition(saltyTopic, 0))); connectorTask.stop(); Assert.assertTrue(connectorTask.awaitStop(CONNECTOR_AWAIT_STOP_TIMEOUT_MS, TimeUnit.MILLISECONDS), "did not shut down on time"); }
@Test public void testIdentityPartitioningEnabled() throws Exception { String yummyTopic = "YummyPizza"; int partitionCount = 10; createTopic(_zkUtils, yummyTopic, partitionCount); Datastream datastream = KafkaMirrorMakerConnectorTestUtils.createDatastream("pizzaStream", _broker, "\\w+Pizza"); datastream.getMetadata() .put(KafkaMirrorMakerDatastreamMetadata.IDENTITY_PARTITIONING_ENABLED, Boolean.TRUE.toString()); DatastreamTaskImpl task = new DatastreamTaskImpl(Collections.singletonList(datastream)); MockDatastreamEventProducer datastreamProducer = new MockDatastreamEventProducer(); task.setEventProducer(datastreamProducer); KafkaMirrorMakerConnectorTask connectorTask = KafkaMirrorMakerConnectorTestUtils.createKafkaMirrorMakerConnectorTask(task); KafkaMirrorMakerConnectorTestUtils.runKafkaMirrorMakerConnectorTask(connectorTask); Set<Integer> expectedPartitionsWithData = new HashSet<>(); for (int i = 0; i < partitionCount; i += 2) { KafkaMirrorMakerConnectorTestUtils.produceEventsToPartition(yummyTopic, i, 1, _kafkaCluster); expectedPartitionsWithData.add(i); } if (!PollUtils.poll(() -> datastreamProducer.getEvents().size() == partitionCount / 2, POLL_PERIOD_MS, POLL_TIMEOUT_MS)) { Assert.fail("did not transfer the msgs within timeout. transferred " + datastreamProducer.getEvents().size()); } List<DatastreamProducerRecord> records = datastreamProducer.getEvents(); for (DatastreamProducerRecord record : records) { String destinationTopic = record.getDestination().get(); record.getPartition().ifPresent(expectedPartitionsWithData::remove); } Assert.assertTrue(expectedPartitionsWithData.isEmpty(), "Partitions did not receive data: " + StringUtils.join(expectedPartitionsWithData, ",")); connectorTask.stop(); Assert.assertTrue(connectorTask.awaitStop(CONNECTOR_AWAIT_STOP_TIMEOUT_MS, TimeUnit.MILLISECONDS), "did not shut down on time"); }
@Test public void testFlushAndCommitDuringGracefulStop() 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(null, null, Duration.ofSeconds(2)); task.setEventProducer(datastreamProducer); KafkaBasedConnectorConfig connectorConfig = new KafkaBasedConnectorConfigBuilder() .setCommitIntervalMillis(600000) .setConsumerFactory(new LiKafkaConsumerFactory()) .build(); KafkaMirrorMakerConnectorTask connectorTask = new KafkaMirrorMakerConnectorTask( connectorConfig, task, "", false, new KafkaGroupIdConstructor(false, "testCluster")); KafkaMirrorMakerConnectorTestUtils.createKafkaMirrorMakerConnectorTask(task); KafkaMirrorMakerConnectorTestUtils.runKafkaMirrorMakerConnectorTask(connectorTask); KafkaMirrorMakerConnectorTestUtils.produceEvents(yummyTopic, 2, _kafkaCluster); if (!PollUtils.poll(() -> datastreamProducer.getEvents().size() == 2, POLL_PERIOD_MS, POLL_TIMEOUT_MS)) { Assert.fail("did not transfer the msgs within timeout. transferred " + datastreamProducer.getEvents().size()); } connectorTask.stop(); Assert.assertTrue(connectorTask.awaitStop(CONNECTOR_AWAIT_STOP_TIMEOUT_MS, TimeUnit.MILLISECONDS), "did not shut down on time"); Assert.assertEquals(datastreamProducer.getNumFlushes(), 1); }
@Test public void testPartitionManagedLockReleaseOnConsumerCloseException() throws Exception { MockDatastreamEventProducer datastreamProducer = new MockDatastreamEventProducer(); Datastream datastream = KafkaMirrorMakerConnectorTestUtils.createDatastream("pizzaStream", _broker, "\\w+Pizza"); DatastreamTaskImpl task = spy(new DatastreamTaskImpl(Collections.singletonList(datastream))); doNothing().when(task).acquire(any()); CountDownLatch releaseCall = new CountDownLatch(1); doAnswer(invocation -> { releaseCall.countDown(); return null; }).when(task).release(); task.setEventProducer(datastreamProducer); CountDownLatch remainingCommitSyncCalls = new CountDownLatch(3); KafkaConsumerFactory<byte[], byte[]> kafkaConsumerFactory = new KafkaConsumerFactoryImpl() { @Override public Consumer<byte[], byte[]> createConsumer(Properties properties) { Consumer<byte[], byte[]> result = spy(super.createConsumer(properties)); doAnswer(invocation -> { remainingCommitSyncCalls.countDown(); return null; }) .when(result).commitSync(anyMapOf(TopicPartition.class, OffsetAndMetadata.class), any(Duration.class)); doThrow(new KafkaException("Throwing close exception")) .when(result).close(); return result; } }; KafkaBasedConnectorConfig connectorConfig = new KafkaBasedConnectorConfigBuilder() .setConsumerFactory(kafkaConsumerFactory) .setCommitIntervalMillis(200) .setEnablePartitionManaged(true) .build(); KafkaMirrorMakerConnectorTask connectorTask = new KafkaMirrorMakerConnectorTask( connectorConfig, task, "", true, new KafkaGroupIdConstructor(false, "test")); KafkaMirrorMakerConnectorTestUtils.runKafkaMirrorMakerConnectorTask(connectorTask); verify(task, times(1)).acquire(any()); verify(task, times(0)).release(); Assert.assertTrue(remainingCommitSyncCalls.await(10, TimeUnit.SECONDS), "Kafka consumer commitSync was not invoked as often as expected"); Assert.assertEquals(datastreamProducer.getNumFlushes(), 0); connectorTask.stop(); Assert.assertTrue(connectorTask.awaitStop(CONNECTOR_AWAIT_STOP_TIMEOUT_MS, TimeUnit.MILLISECONDS), "did not shut down on time"); Assert.assertEquals(datastreamProducer.getNumFlushes(), 1); verify(task, times(1)).acquire(any()); Assert.assertTrue(releaseCall.await(10, TimeUnit.SECONDS), "DatastreamTask never released"); }
@Test public void testRegularCommitWithFlushlessProducer() throws Exception { MockDatastreamEventProducer datastreamProducer = new MockDatastreamEventProducer(); Datastream datastream = KafkaMirrorMakerConnectorTestUtils.createDatastream("pizzaStream", _broker, "\\w+Pizza"); DatastreamTaskImpl task = new DatastreamTaskImpl(Collections.singletonList(datastream)); task.setEventProducer(datastreamProducer); CountDownLatch remainingCommitSyncCalls = new CountDownLatch(3); KafkaConsumerFactory<byte[], byte[]> kafkaConsumerFactory = new KafkaConsumerFactoryImpl() { @Override public Consumer<byte[], byte[]> createConsumer(Properties properties) { Consumer<byte[], byte[]> result = spy(super.createConsumer(properties)); doAnswer(invocation -> { remainingCommitSyncCalls.countDown(); return null; }) .when(result).commitSync(anyMapOf(TopicPartition.class, OffsetAndMetadata.class), any(Duration.class)); return result; } }; KafkaBasedConnectorConfig connectorConfig = new KafkaBasedConnectorConfigBuilder() .setConsumerFactory(kafkaConsumerFactory) .setCommitIntervalMillis(200) .build(); KafkaMirrorMakerConnectorTask connectorTask = new KafkaMirrorMakerConnectorTask( connectorConfig, task, "", true, new KafkaGroupIdConstructor(false, "test")); KafkaMirrorMakerConnectorTestUtils.runKafkaMirrorMakerConnectorTask(connectorTask); Assert.assertTrue(remainingCommitSyncCalls.await(10, TimeUnit.SECONDS), "Kafka consumer commitSync was not invoked as often as expected"); Assert.assertEquals(datastreamProducer.getNumFlushes(), 0); connectorTask.stop(); Assert.assertTrue(connectorTask.awaitStop(CONNECTOR_AWAIT_STOP_TIMEOUT_MS, TimeUnit.MILLISECONDS), "did not shut down on time"); Assert.assertEquals(datastreamProducer.getNumFlushes(), 1); }
@Test public void testAutoOffsetResetConfigOverride() throws Exception { String yummyTopic = "YummyPizza"; createTopic(_zkUtils, yummyTopic); Datastream datastreamLatest = KafkaMirrorMakerConnectorTestUtils.createDatastream("pizzaStream", _broker, "\\w+Pizza"); datastreamLatest.getMetadata().put(KafkaDatastreamMetadataConstants.CONSUMER_OFFSET_RESET_STRATEGY, AbstractKafkaBasedConnectorTask.CONSUMER_AUTO_OFFSET_RESET_CONFIG_LATEST); datastreamLatest.getMetadata().put(DatastreamMetadataConstants.GROUP_ID, "groupIdForDatastreamLatest"); Datastream datastreamEarliest = KafkaMirrorMakerConnectorTestUtils.createDatastream("pizzaStream", _broker, "\\w+Pizza"); datastreamEarliest.getMetadata().put(KafkaDatastreamMetadataConstants.CONSUMER_OFFSET_RESET_STRATEGY, AbstractKafkaBasedConnectorTask.CONSUMER_AUTO_OFFSET_RESET_CONFIG_EARLIEST); datastreamEarliest.getMetadata().put(DatastreamMetadataConstants.GROUP_ID, "groupIdForDatastreamEarliest"); KafkaMirrorMakerConnectorTestUtils.produceEvents(yummyTopic, 5, _kafkaCluster); DatastreamTaskImpl taskLatest = new DatastreamTaskImpl(Collections.singletonList(datastreamLatest)); MockDatastreamEventProducer datastreamProducerLatest = new MockDatastreamEventProducer(); taskLatest.setEventProducer(datastreamProducerLatest); DatastreamTaskImpl taskEarliest = new DatastreamTaskImpl(Collections.singletonList(datastreamEarliest)); MockDatastreamEventProducer datastreamProducerEarliest = new MockDatastreamEventProducer(); taskEarliest.setEventProducer(datastreamProducerEarliest); KafkaMirrorMakerConnectorTask connectorTaskLatest = KafkaMirrorMakerConnectorTestUtils.createKafkaMirrorMakerConnectorTask(taskLatest); KafkaMirrorMakerConnectorTestUtils.runKafkaMirrorMakerConnectorTask(connectorTaskLatest); KafkaMirrorMakerConnectorTask connectorTaskEarliest = KafkaMirrorMakerConnectorTestUtils.createKafkaMirrorMakerConnectorTask(taskEarliest); KafkaMirrorMakerConnectorTestUtils.runKafkaMirrorMakerConnectorTask(connectorTaskEarliest); Assert.assertTrue(PollUtils.poll( () -> datastreamProducerEarliest.getEvents().size() == 5 && datastreamProducerLatest.getEvents().size() == 0, POLL_PERIOD_MS, POLL_TIMEOUT_MS), "Datastream configured with offset reset latest should have gotten " + "0 events, and datastream configured with earliest should have gotten 5 events"); KafkaMirrorMakerConnectorTestUtils.produceEvents(yummyTopic, 5, _kafkaCluster); Assert.assertTrue(PollUtils.poll( () -> datastreamProducerEarliest.getEvents().size() == 10 && datastreamProducerLatest.getEvents().size() == 5, POLL_PERIOD_MS, POLL_TIMEOUT_MS), "Datastream configured with offset reset latest should have gotten " + "5 events, and datastream configured with earliest should have gotten 10 events"); connectorTaskLatest.stop(); Assert.assertTrue(connectorTaskLatest.awaitStop(CONNECTOR_AWAIT_STOP_TIMEOUT_MS, TimeUnit.MILLISECONDS), "did not shut down on time"); connectorTaskEarliest.stop(); Assert.assertTrue(connectorTaskEarliest.awaitStop(CONNECTOR_AWAIT_STOP_TIMEOUT_MS, TimeUnit.MILLISECONDS), "did not shut down on time"); }
@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"); }
@Test public void testAutoPauseAndResumeOnSendFailure() 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); if (!PollUtils.poll(() -> connectorTask.getAutoPausedSourcePartitions().contains(new TopicPartition(yummyTopic, 0)), 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"); validatePausedPartitionsMetrics("KafkaMirrorMakerConnectorTask", datastream.getName(), 1, 0, 0); datastreamProducer.setSendFailCondition((r) -> false); Assert.assertTrue( PollUtils.poll(() -> connectorTask.getAutoPausedSourcePartitions().isEmpty(), POLL_PERIOD_MS, POLL_TIMEOUT_MS), "Partition that was auto-paused did not auto-resume."); if (!PollUtils.poll(() -> datastreamProducer.getEvents().size() == 7, POLL_PERIOD_MS, POLL_TIMEOUT_MS)) { Assert.fail("datastream producer " + datastreamProducer.getEvents().size()); } validatePausedPartitionsMetrics("KafkaMirrorMakerConnectorTask", datastream.getName(), 0, 0, 0); connectorTask.stop(); Assert.assertTrue(connectorTask.awaitStop(CONNECTOR_AWAIT_STOP_TIMEOUT_MS, TimeUnit.MILLISECONDS), "did not shut down on time"); }
@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"); }