Как обновить свойства SSL клиента KAFKA с помощью SSLBundle ⇐ JAVA

[Anonymous]

Я пытаюсь опубликовать сообщения в кластер Kafka, который защищен SSL. < /p>
У меня есть правильные сертификаты на стороне приложения. Я проверил, что могу потреблять из этого кластера Kafka, используя их. Кроме того, тема существует, мне не нужно ее создавать. Мне просто нужно опубликовать его.

Код: Выделить всё

[nio-8080-exec-1] [4af24028c1d624fb61198f30b483900f-d67a602cf7c48e65] o.a.k.clients.admin.AdminClientConfig    : AdminClientConfig values:
auto.include.jmx.reporter = true
bootstrap.controllers = []
bootstrap.servers = [mykafka.com:9092]
client.dns.lookup = use_all_dns_ips
client.id =
socket.connection.setup.timeout.max.ms = 30000
socket.connection.setup.timeout.ms = 10000
ssl.cipher.suites = null
ssl.enabled.protocols = [TLSv1.2, TLSv1.3]
ssl.endpoint.identification.algorithm = https
ssl.engine.factory.class = null
ssl.key.password = null
ssl.keymanager.algorithm = SunX509
ssl.keystore.certificate.chain = null
ssl.keystore.key = null
ssl.keystore.location = null
ssl.keystore.password = null
ssl.keystore.type = JKS
ssl.protocol = TLSv1.3
ssl.provider = null
ssl.secure.random.implementation = null
ssl.trustmanager.algorithm = PKIX
ssl.truststore.certificates = null
ssl.truststore.location = null
ssl.truststore.password = null
ssl.truststore.type = JKS

[nio-8080-exec-1] [4af24028c1d624fb61198f30b483900f-d67a602cf7c48e65] o.a.kafka.common.utils.AppInfoParser     : Kafka version: 3.9.1
[nio-8080-exec-1] [4af24028c1d624fb61198f30b483900f-d67a602cf7c48e65] o.a.kafka.common.utils.AppInfoParser     : Kafka commitId: f745dfdcee2b9851
[nio-8080-exec-1] [4af24028c1d624fb61198f30b483900f-d67a602cf7c48e65] o.a.kafka.common.utils.AppInfoParser     : Kafka startTimeMs: 1749465196889
[| adminclient-1] [                                                 ] org.apache.kafka.clients.NetworkClient   : [AdminClient clientId=adminclient-1] Node -1 disconnected.
[| adminclient-1] [                                                 ] org.apache.kafka.clients.NetworkClient   : [AdminClient clientId=adminclient-1] Cancelled in-flight API_VERSIONS request with correlation id 0 due to node -1 being disconnected (elapsed time since creation: 413ms, elapsed time since send: 413ms, throttle time: 0ms, request timeout: 3600000ms)
[| adminclient-1] [                                                 ] org.apache.kafka.clients.NetworkClient   : [AdminClient clientId=adminclient-1] Node -1 disconnected.
[| adminclient-1] [                                                 ] org.apache.kafka.clients.NetworkClient   : [AdminClient clientId=adminclient-1] Cancelled in-flight API_VERSIONS request with correlation id 1 due to node -1 being disconnected (elapsed time since creation: 247ms, elapsed time since send: 247ms, throttle time: 0ms, request timeout: 3600000ms)

(почему?) />

Код: Выделить всё

spring:
application:
name: question
kafka:
properties:
security.protocol: SSL
ssl:
key-password: abc
keystore-location: 'file:///Users/me/keystore.p12'
keystore-password: abc
truststore-location: 'file:///Users/me/truststore.p12'
truststore-password: abc
< /code>
Независимо от того, я хотел бы обновить его, используя программный подход, это то, что я попробовал для производителя.  < /p>
@Configuration
public class KafkaProducerConfiguration {

@Bean
public KafkaTemplate kafkaTemplate(final MeterRegistry meterRegistry, final SslBundles sslBundles) {
final Map properties = new ConcurrentHashMap();
properties.put("security.protocol", "SSL");
properties.put(SslConfigs.SSL_ENGINE_FACTORY_CLASS_CONFIG, SslBundleSslEngineFactory.class.getName());
properties.put(SslBundle.class.getName(), sslBundles.getBundle("my-bundle"));
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "mykafka.com:9092");
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);

final DefaultKafkaProducerFactory defaultKafkaProducerFactory = new DefaultKafkaProducerFactory(properties);
defaultKafkaProducerFactory.addListener(new MicrometerProducerListener(meterRegistry));
KafkaTemplate kafkaTemplate = new KafkaTemplate(defaultKafkaProducerFactory);
kafkaTemplate.setObservationEnabled(true);
kafkaTemplate.setObservationConvention(new KafkaTemplateObservation.DefaultKafkaTemplateObservationConvention());
return kafkaTemplate;
}

}

Я хотел бы придумать какой -то аналогичный подход для клиента администратора, но я не вижу поминтурема.>

Подробнее здесь: https://stackoverflow.com/questions/796 ... -sslbundle