@Test public void testConsumerProperties() { Properties overrides = new Properties(); String groupId = "groupId"; KafkaConnectionString connectionString = KafkaConnectionString.valueOf("kafka: Properties actual = KafkaConnectorTask.getKafkaConsumerProperties(overrides, groupId, connectionString); Properties expected = new Properties(); expected.put("auto.offset.reset", "none"); expected.put("bootstrap.servers", "MyBroker:10251"); expected.put("enable.auto.commit", "false"); expected.put("group.id", "groupId"); expected.put("security.protocol", "PLAINTEXT"); Assert.assertEquals(actual, expected); }
@VisibleForTesting static Properties getKafkaConsumerProperties(Properties consumerProps, String groupId, KafkaConnectionString connectionString) { StringJoiner csv = new StringJoiner(","); connectionString.getBrokers().forEach(broker -> csv.add(broker.toString())); String bootstrapValue = csv.toString(); Properties props = new Properties(); props.putAll(consumerProps); props.putIfAbsent(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapValue); props.putIfAbsent(ConsumerConfig.GROUP_ID_CONFIG, groupId); props.putIfAbsent(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); props.putIfAbsent(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "none"); props.putIfAbsent(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, connectionString.isSecure() ? "SSL" : "PLAINTEXT"); return props; }
KafkaConnectorTask extends AbstractKafkaBasedConnectorTask { @VisibleForTesting static Properties getKafkaConsumerProperties(Properties consumerProps, String groupId, KafkaConnectionString connectionString) { StringJoiner csv = new StringJoiner(","); connectionString.getBrokers().forEach(broker -> csv.add(broker.toString())); String bootstrapValue = csv.toString(); Properties props = new Properties(); props.putAll(consumerProps); props.putIfAbsent(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapValue); props.putIfAbsent(ConsumerConfig.GROUP_ID_CONFIG, groupId); props.putIfAbsent(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); props.putIfAbsent(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "none"); props.putIfAbsent(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, connectionString.isSecure() ? "SSL" : "PLAINTEXT"); return props; } }
KafkaConnectorTask extends AbstractKafkaBasedConnectorTask { @VisibleForTesting static Properties getKafkaConsumerProperties(Properties consumerProps, String groupId, KafkaConnectionString connectionString) { StringJoiner csv = new StringJoiner(","); connectionString.getBrokers().forEach(broker -> csv.add(broker.toString())); String bootstrapValue = csv.toString(); Properties props = new Properties(); props.putAll(consumerProps); props.putIfAbsent(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapValue); props.putIfAbsent(ConsumerConfig.GROUP_ID_CONFIG, groupId); props.putIfAbsent(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); props.putIfAbsent(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "none"); props.putIfAbsent(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, connectionString.isSecure() ? "SSL" : "PLAINTEXT"); return props; } KafkaConnectorTask(KafkaBasedConnectorConfig config, DatastreamTask task, String connectorName,
      GroupIdConstructor groupIdConstructor); }
KafkaConnectorTask extends AbstractKafkaBasedConnectorTask { @VisibleForTesting static Properties getKafkaConsumerProperties(Properties consumerProps, String groupId, KafkaConnectionString connectionString) { StringJoiner csv = new StringJoiner(","); connectionString.getBrokers().forEach(broker -> csv.add(broker.toString())); String bootstrapValue = csv.toString(); Properties props = new Properties(); props.putAll(consumerProps); props.putIfAbsent(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapValue); props.putIfAbsent(ConsumerConfig.GROUP_ID_CONFIG, groupId); props.putIfAbsent(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); props.putIfAbsent(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "none"); props.putIfAbsent(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, connectionString.isSecure() ? "SSL" : "PLAINTEXT"); return props; } KafkaConnectorTask(KafkaBasedConnectorConfig config, DatastreamTask task, String connectorName,
      GroupIdConstructor groupIdConstructor); static Consumer<?, ?> createConsumer(KafkaConsumerFactory<?, ?> consumerFactory, Properties consumerProps,
      String groupId, KafkaConnectionString connectionString); static List<BrooklinMetricInfo> getMetricInfos(String connectorName); }
KafkaConnectorTask extends AbstractKafkaBasedConnectorTask { @VisibleForTesting static Properties getKafkaConsumerProperties(Properties consumerProps, String groupId, KafkaConnectionString connectionString) { StringJoiner csv = new StringJoiner(","); connectionString.getBrokers().forEach(broker -> csv.add(broker.toString())); String bootstrapValue = csv.toString(); Properties props = new Properties(); props.putAll(consumerProps); props.putIfAbsent(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapValue); props.putIfAbsent(ConsumerConfig.GROUP_ID_CONFIG, groupId); props.putIfAbsent(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); props.putIfAbsent(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "none"); props.putIfAbsent(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, connectionString.isSecure() ? "SSL" : "PLAINTEXT"); return props; } KafkaConnectorTask(KafkaBasedConnectorConfig config, DatastreamTask task, String connectorName,
      GroupIdConstructor groupIdConstructor); static Consumer<?, ?> createConsumer(KafkaConsumerFactory<?, ?> consumerFactory, Properties consumerProps,
      String groupId, KafkaConnectionString connectionString); static List<BrooklinMetricInfo> getMetricInfos(String connectorName);  }