KafkaConnector extends AbstractKafkaConnector { @Override public void initializeDatastream(Datastream stream, List<Datastream> allDatastreams) throws DatastreamValidationException { LOG.info("Initialize datastream {}", stream); DatastreamSource source = stream.getSource(); String connectionString = source.getConnectionString(); try { KafkaConnectionString parsed = KafkaConnectionString.valueOf(connectionString); source.setConnectionString(parsed.toString()); if (stream.hasDestination()) { if (!stream.getDestination().hasKeySerDe() && !StringUtils.isBlank(_config.getDefaultKeySerde())) { stream.getDestination().setKeySerDe(_config.getDefaultKeySerde()); } if (!stream.getDestination().hasPayloadSerDe() && !StringUtils.isBlank(_config.getDefaultValueSerde())) { stream.getDestination().setPayloadSerDe(_config.getDefaultValueSerde()); } } if (!isWhiteListedCluster(parsed)) { String msg = String.format("Kafka connector is not white-listed for the cluster %s. Current white-listed clusters %s.", connectionString, _whiteListedBrokers); LOG.error(msg); throw new DatastreamValidationException(msg); } try (Consumer<?, ?> consumer = KafkaConnectorTask.createConsumer(_config.getConsumerFactory(), _config.getConsumerProps(), "partitionFinder", parsed)) { List<PartitionInfo> partitionInfos = consumer.partitionsFor(parsed.getTopicName()); if (partitionInfos == null) { throw new DatastreamValidationException( "Can't get partition info from kafka. Very likely that auto topic creation " + "is disabled on the broker and the topic doesn't exist: " + parsed.getTopicName()); } int numPartitions = partitionInfos.size(); if (!source.hasPartitions()) { LOG.info("Kafka source {} has {} partitions.", parsed, numPartitions); source.setPartitions(numPartitions); } else { if (source.getPartitions() != numPartitions) { String msg = String.format("Source is configured with %d partitions, But the topic %s actually has %d partitions", source.getPartitions(), parsed.getTopicName(), numPartitions); LOG.error(msg); throw new DatastreamValidationException(msg); } } if (stream.getMetadata().containsKey(DatastreamMetadataConstants.START_POSITION)) { String json = stream.getMetadata().get(DatastreamMetadataConstants.START_POSITION); Map<Integer, Long> offsetMap = JsonUtils.fromJson(json, new TypeReference<Map<Integer, Long>>() { }); if (offsetMap.size() != numPartitions || IntStream.range(0, numPartitions) .anyMatch(x -> !offsetMap.containsKey(x))) { String msg = String.format("Missing partitions starting offset for datastream %s, json value %s", stream, json); LOG.warn(msg); throw new DatastreamValidationException(msg); } } } } catch (DatastreamValidationException e) { throw e; } catch (Exception e) { LOG.error("Initialization threw an exception.", e); throw new DatastreamValidationException(e); } } KafkaConnector(String connectorName, Properties config, String clusterName); @Override void initializeDatastream(Datastream stream, List<Datastream> allDatastreams); @Override List<BrooklinMetricInfo> getMetricInfos(); @Override void postDatastreamInitialize(Datastream datastream, List<Datastream> allDatastreams); static final String CONFIG_WHITE_LISTED_CLUSTERS; }
@Test public void testConnectorWithStartPosition() throws UnsupportedEncodingException, DatastreamValidationException { String topicName = "testConnectorWithStartPosition"; TestKafkaConnectorTask.produceEvents(_kafkaCluster, _zkUtils, topicName, 0, 100); long ts = System.currentTimeMillis(); TestKafkaConnectorTask.produceEvents(_kafkaCluster, _zkUtils, topicName, 100, 100); Datastream ds = createDatastream("testConnectorPopulatesPartitions", topicName); Map<Integer, Long> offsets = Collections.singletonMap(0, 100L); KafkaConnector connector = new KafkaConnector("test", getDefaultConfig(null), "testCluster"); ds.getMetadata().put(DatastreamMetadataConstants.START_POSITION, JsonUtils.toJson(offsets)); connector.initializeDatastream(ds, Collections.emptyList()); }
@Test(expectedExceptions = DatastreamValidationException.class) public void testInitializeDatastreamWithNonexistentTopic() throws DatastreamValidationException { String topicName = "testInitializeDatastreamWithNonexistentTopic"; Datastream ds = createDatastream("testInitializeDatastreamWithNonexistentTopic", topicName); KafkaConnector connector = new KafkaConnector("test", getDefaultConfig(null), "testCluster"); connector.initializeDatastream(ds, Collections.emptyList()); }
@Test public void testPopulatingDefaultSerde() throws Exception { String topicName = "testPopulatingDefaultSerde"; TestKafkaConnectorTask.produceEvents(_kafkaCluster, _zkUtils, topicName, 0, 100); TestKafkaConnectorTask.produceEvents(_kafkaCluster, _zkUtils, topicName, 100, 100); Datastream ds = createDatastream("testPopulatingDefaultSerde", topicName); KafkaConnector connector = new KafkaConnector("test", getDefaultConfig(null), "testCluster"); connector.initializeDatastream(ds, Collections.emptyList()); Assert.assertTrue(ds.getDestination().hasKeySerDe()); Assert.assertEquals(ds.getDestination().getKeySerDe(), "keySerde"); Assert.assertTrue(ds.getDestination().hasPayloadSerDe()); Assert.assertEquals(ds.getDestination().getPayloadSerDe(), "valueSerde"); }
@Test public void testConnectorPopulatesPartitions() throws UnsupportedEncodingException, DatastreamValidationException { String topicName = "testConnectorPopulatesPartitions"; TestKafkaConnectorTask.produceEvents(_kafkaCluster, _zkUtils, topicName, 0, 10); Datastream ds = createDatastream("testConnectorPopulatesPartitions", topicName); KafkaConnector connector = new KafkaConnector("test", getDefaultConfig(null), "testCluster"); connector.initializeDatastream(ds, Collections.emptyList()); Assert.assertEquals(ds.getSource().getPartitions().intValue(), 1); }
@Test(expectedExceptions = DatastreamValidationException.class) public void testConnectorValidatesWhitelistedBroker() throws DatastreamValidationException { String topicName = "testConnectorValidatesWhitelistedBroker"; Datastream ds = createDatastream("testConnectorPopulatesPartitions", topicName); Properties override = new Properties(); override.put(KafkaConnector.CONFIG_WHITE_LISTED_CLUSTERS, "randomBroker:2546"); KafkaConnector connector = new KafkaConnector("test", getDefaultConfig(override), "testCluster"); connector.initializeDatastream(ds, Collections.emptyList()); }