### Question:
KafkaConnectionCustomizer implements ComponentProxyCustomizer, CamelContextAware { @Override public void customize(ComponentProxyComponent component, Map<String, Object> options) { KafkaConfiguration configuration = new KafkaConfiguration(); if (ConnectorOptions.extractOption(options, CERTIFICATE_OPTION) != null) { LOG.info("Setting SSLContextParameters configuration as a self-signed certificate was provided"); SSLContextParameters sslContextParameters = createSSLContextParameters( ConnectorOptions.extractOption(options, CERTIFICATE_OPTION)); configuration.setSslContextParameters(sslContextParameters); configuration.setSecurityProtocol("SSL"); configuration.setSslEndpointAlgorithm(""); } String extraOptions = options.getOrDefault("extraOptions", "[]").toString(); try { @SuppressWarnings("unchecked") List<Map<String, String>> attributes = MAPPER.readValue(extraOptions, List.class); for (Map<String, String> attribute : attributes) { final String key = attribute.get("key").trim(); if (!key.isEmpty()) { final String value = attribute.get("value"); final String property = getCanonicalPropertyName(key); options.put(property, value); configuration.getAdditionalProperties().put("additionalProperties." + property, value); } } PropertyBindingSupport.bindProperties(this.getCamelContext(), configuration, options); } catch (JsonProcessingException e) { LOG.error(e.getMessage(), e); } options.put("configuration", configuration); } KafkaConnectionCustomizer(CamelContext context); @Override CamelContext getCamelContext(); @Override final void setCamelContext(CamelContext camelContext); @Override void customize(ComponentProxyComponent component, Map<String, Object> options); static final ObjectMapper MAPPER; }### Answer:
@Test public void shouldAssertIfSelfSignedCertificate() { KafkaConnectionCustomizer kafkaConnectionCustomizer = new KafkaConnectionCustomizer(new DefaultCamelContext()); Map<String, Object> options = new HashMap<>(); options.put("brokers", "test:9092"); options.put("brokerCertificate", TEST_CERT); kafkaConnectionCustomizer.customize(null, options); assertThat(options).containsKey("configuration"); KafkaConfiguration kafkaConfiguration = (KafkaConfiguration) options.get("configuration"); assertThat(kafkaConfiguration.getSslContextParameters()).isNotNull(); assertThat(kafkaConfiguration.getSecurityProtocol()).isEqualTo("SSL"); assertThat(kafkaConfiguration.getSslEndpointAlgorithm()).isEqualTo(""); }
@Test public void testExtraOptions() { KafkaConnectionCustomizer kafkaConnectionCustomizer = new KafkaConnectionCustomizer(new DefaultCamelContext()); Map<String, Object> options = new HashMap<>(); options.put("brokers", "test:9092"); options.put("extraOptions", "[{\"key\":\"AAA\",\"value\":\"BBB\"}," + " {\"key\":\"autoOffsetReset\"," + "\"value\":\"earliest\"}," + "{\"key\":\"checkCrcs\",\"value\":\"false\"}, " + "{\"key\":\"auto.commit.interval.ms\",\"value\":\"5\"}]"); kafkaConnectionCustomizer.customize(null, options); assertThat(options).containsKey("configuration"); KafkaConfiguration kafkaConfiguration = (KafkaConfiguration) options.get("configuration"); assertThat(kafkaConfiguration.getAdditionalProperties()).containsEntry("additionalProperties.AAA", "BBB"); assertThat(kafkaConfiguration.getAutoOffsetReset()).isEqualTo("earliest"); assertThat(kafkaConfiguration.getCheckCrcs()).isEqualTo(false); assertThat(kafkaConfiguration.getAutoCommitIntervalMs()).isEqualTo(5); }
@Test public void testExtraOptionsEmpty() { KafkaConnectionCustomizer kafkaConnectionCustomizer = new KafkaConnectionCustomizer(new DefaultCamelContext()); Map<String, Object> options = new HashMap<>(); options.put("brokers", "test:9092"); options.put("extraOptions", "[]"); kafkaConnectionCustomizer.customize(null, options); assertThat(options).containsKey("configuration"); KafkaConfiguration kafkaConfiguration = (KafkaConfiguration) options.get("configuration"); assertThat(kafkaConfiguration.getAdditionalProperties()).isEmpty(); }