@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(""); }
@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 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 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); }
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); }
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; }