Привет, Хабр! Сегодня я расскажу о своем опыте создания сервиса для отправки сообщений в Apache Kafka с использованием Spring Boot, аутентификацией SASL/Kerberos и применением Avro Schema Registry для продюсера. В процессе разработки я столкнулся с рядом проблем, решение которых потребовало усилий и времени. Надеюсь, мой опыт будет полезен разработчикам, которым предстоит решать подобные задачи.
Ведение
По техзаданию необходимо было создать сервис, который в зависимости от топика отправлял бы сообщения или на один инстанс Kafka (строку, с простой авторизацией с помощью SSL), или на другой, но уже с сериализацией и аутентификацией через Kerberos.
Обмен данными с обоими серверами Kafka использует шифрование с помощью сертификатов, второй использует дополнительную аутентификацию клиента с помощью Kerberos.
Приложение деплоится на Томкат.
Немного теории
Apache Kafka — популярная платформа потоковой передачи данных, часто используемая для создания высоконагруженных распределенных систем. Для обеспечения безопасности могут применятся механизмы аутентификации SASL/Kerberos. Avro Schema Registry используется для управления схемами данных и обеспечивает совместимость версий сообщений и валидации отправляемых\принимаемых данных.
Kerberos — это протокол аутентификации, разработанный для обеспечения безопасного доступа к сети. Он работает на основе системы тикетов, предоставляемых центральным сервером аутентификации (KDC — Key Distribution Center).
SSL (Secure Sockets Layer) используется для шифрования данных и аутентификации сервера и/или клиента. Чтобы настроить SSL для соединения с Kafka с помощью JKS (Java KeyStore), необходимо создать и настроить truststore и keystore файлы. Truststore содержит сертификаты доверенных удостоверяющих центров (CA), а keystore содержит закрытые ключи и сертификаты для идентификации сервера или клиента. В Kafka клиентской конфигурации необходимо указать местоположения этих файлов и их пароли.
Настройка Spring Boot и зависимостей
Первым шагом добавляем необходимые зависимости в pom.xml:
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
<groupId>io.confluent</groupId>
<artifactId>kafka-avro-serializer</artifactId>
<version>6.2.0</version>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.8.0</version>
</dependency>
</dependencies>Настройка SSL
В конфигурации соединения необходимо указать путь к jks файлу и пароля к нему. В примере ниже это producerRetailFactory.
Настройка Kerberos
Настройка Kerberos требует наличия конфигурационного файла krb5.conf и указания пути к keytab. Файл krb5.conf должен содержать информацию о KDC (Key Distribution Center). В примере ниже это producerMainFactory.
[libdefaults]
default_realm = TEST_DOMAIN.TRUS.TCOMPANY
[realms]
TEST_DOMAIN.TRUS.TCOMPANY = {
kdc = DC00.TEST_DOMAIN.trus.tCOMPANY
admin_server = S000.TEST_DOMAIN.trus.tCOMPANY
kdc = DC01.TEST_DOMAIN.TRUS.TCOMPANY
}
[domain_realm]
.TEST_DOMAIN.trus.tCOMPANY = TEST_DOMAIN.TRUS.TCOMPANY
TEST_DOMAIN.trus.tCOMPANY = TEST_DOMAIN.TRUS.TCOMPANYkeytab это двоичный файл содержащий пары Kerberos принципалов и их ключей (полученных с использованием Kerberos пароля). Эти файлы используются для аутентификации, без ввода пароля.
Обычно эти данные предоставляются администраторами сервисов.
Конфигурация для двух серверов Kafka
Так как у нас два разных брокера Kafka, требующих разные типы аутентификации, мы создадим специальный класс для конфигурирования. Один будет использовать SSL для аутентификации, другой — SSL, SASL/Kerberos.
Важно, в классе конфигурации, путь к
krb5.confдолжен быть отдельно указан (иначе будет использован системный) и указывать его нужно не через props (как остальные параметры), а через установку переменной окружения.
System.setProperty("java.security.krb5.conf", krb5ConfFile);@Configuration
@EnableKafka
@Slf4j
public class KafkaConfig {
private final Environment environment;
private final JMServiceDao jmServiceDao;
public KafkaConfig(Environment environment, JMServiceDao jmServiceDao) {
this.environment = environment;
this.jmServiceDao = jmServiceDao;
}
@Bean
public ProducerFactory<String, Object> producerRetailFactory() {
Map<String, Object> props = new HashMap<>();
String hostname = Optional.ofNullable(System.getenv("HOSTNAME")).filter(Predicate.not(String::isBlank)).orElse(environment.getProperty("COMPUTERNAME", "Localhost"));
props.put(CLIENT_ID_CONFIG, hostname + "_" + environment.getProperty("spring.application.name"));
props.put(BOOTSTRAP_SERVERS_CONFIG, environment.getProperty("KafkaRetailBrokerHost", jmServiceDao.getKafkaRetailBrokerHost()));
props.put(KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
props.put(VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
props.put("security.protocol", "SSL");
props.put("ssl.enabled.protocols", "TLSv1.2");
props.put("ssl.truststore.location", environment.getProperty("KafkaRetailSSLKeyStoreLocation" + "kafka.retail.truststore.jks",
environment.getProperty("catalina.base") +
jmServiceDao.getKafkaRetailSSLKeyStoreLocation()));
props.put("ssl.truststore.password", environment.getProperty("KafkaRetailSSLKeyStorePwd", jmServiceDao.getKafkaRetailSSLKeyStorePassword()));
return new DefaultKafkaProducerFactory<>(props);
}
@Bean
public ProducerFactory<String, Object> producerMainFactory() {
// важно! путь к keyTab должен быть экраннированным для windows path
String keyTabFile = Paths.get(environment.getProperty("catalina.base") + jmServiceDao.getKafkaMainKeytabLocation()).toAbsolutePath().normalize().toString().replace("\\", "/");
String krb5ConfFile = Paths.get(environment.getProperty("catalina.base") + jmServiceDao.getKafkaMainKrb5ConfLocation()).toAbsolutePath().normalize().toString().replace("\\", "/");
String jksFile = Paths.get(environment.getProperty("KafkaMainSSLKeyStoreLocation" + "kafka.retail.truststore.jks",
environment.getProperty("catalina.base") + jmServiceDao.getKafkaMainSSLKeyStoreLocation())).toAbsolutePath().normalize().toString().replace("\\", "/");
Map<String, Object> props = new HashMap<>();
String hostname = Optional.ofNullable(environment.getProperty("HOSTNAME")).filter(Predicate.not(String::isBlank)).orElse(environment.getProperty("COMPUTERNAME", "Localhost"));
props.put(CLIENT_ID_CONFIG, hostname + "_" + environment.getProperty("spring.application.name"));
props.put(BOOTSTRAP_SERVERS_CONFIG, environment.getProperty("KafkaMainBrokerHost", jmServiceDao.getKafkaMainBrokerHost()));
props.put(KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
props.put(VALUE_SERIALIZER_CLASS_CONFIG, "io.confluent.kafka.serializers.KafkaAvroSerializer");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "io.confluent.kafka.serializers.KafkaAvroDeserializer");
props.put("specific.avro.reader", "true");
props.put("sasl.mechanism", "GSSAPI");
props.put("sasl.jaas.config", """
com.sun.security.auth.module.Krb5LoginModule required serviceName="kafka"
useKeyTab=true storeKey=true keyTab="%s" principal="%s";
""".formatted(keyTabFile, jmServiceDao.getKafkaMainPrincipal()));
props.put("session.timeout.ms", "45000");
props.put("schema.registry.url", jmServiceDao.getKafkaMainSchemaRegistryUrl());
// параметр устанавливается не через props
System.setProperty("java.security.krb5.conf", krb5ConfFile);
props.put("security.protocol", "SASL_SSL");
props.put("ssl.enabled.protocols", "TLSv1.2,TLSv1.1,TLSv1");
props.put("ssl.truststore.location", environment.getProperty("KafkaMainSSLKeyStoreLocation" + "kafka.main.truststore.jks", jksFile));
props.put("ssl.truststore.password", environment.getProperty("KafkaMainSSLKeyStorePwd", jmServiceDao.getKafkaMainSSLKeyStorePassword()));
//log.debug("kafka props {}", props);
//log.debug("system env {}", System.getProperties());
return new DefaultKafkaProducerFactory<>(props);
}
@Bean
public KafkaTemplate<String, Object> kafkaTemplateForRetail() {
return new KafkaTemplate<>(producerRetailFactory());
}
@Bean
public KafkaTemplate<String, Object> kafkaTemplateForMain() {
return new KafkaTemplate<>(producerMainFactory());
}
}
В этом классе необходимо обратить внимание (кроме обязательных параметров), на параметр sasl.jaas.config и здесь я столкнулся с первой проблемой, оказалось, что нигде нет проверки на правильность пути к keyTab файлу. Разработка велась в ОС Windows, и windows путь, например C:\ra_projects\Java\apache-tomcat-10.1.10\conf\keystore\tkafka.keytab из-за отсутствия экранирования слешей (путь хранится в БД), преобразуется во внутренностях класса связанного с авторизацией, в C:a_projectsJava�pache-tomcat-10.1.10confkeystore kafka.keytab, в Exception я получал только, что авторизация не удалась:
Caused by: org.apache.kafka.common.KafkaException: javax.security.auth.login.LoginException: Could not login: the client is being asked for a password, but the Kafka client code does not currently support obtaining a password from the user. not available to garner authentication information from the user
И только в режиме Debug я увидел кривой путь к файлу
keyTab. Добавил экранирование. Так же завел issue.
Вторая проблема, я рандомно получал ошибку:
javax.security.sasl.SaslException: An error: (java.security.PrivilegedActionException: javax.security.sasl.SaslException: GSS initiate failed [Caused by GSSException: No valid credentials provided (Mechanism level: Server not found in Kerberos database (7) - LOOKING_UP_SERVER)]) occurred when evaluating SASL token received from the Kafka Broker. Kafka Client will retry.
Оказалось, что мне предоставили неправильный файл
krb5.conf, содержащий неправильные пути к KDC доменам. А в последствии, что распарсенный файлkrb5.confхранится кэшированным, и даже если его содержимое поменять, он не будет перечитан до перезапуска Tomcat сервера, на который деплоилось мое приложение.
Сериализация данных
В конфигурационном классе KafkaConfig необходимо указать домен к хранилище схем:
props.put("schema.registry.url", jmServiceDao.getKafkaMainSchemaRegistryUrl());Для отправки данных в Kafka был написан класс, который использует две конфигурации в зависимости от данных, которые были получены для отправки:
@Service
@Slf4j
public class JMServiceImpl implements JMService {
private enum KafkaType {
KAFKA_RETAIL,
KAFKA_MAIN
}
private final JMServiceDao jmServiceDao;
// авторизация через SSL, тут продюсер для отправки "сырых" данных
private final KafkaTemplate<String, Object> kafkaTemplateForRetail;
// авторизация через SSL,Kerberos, тут продюсер для отправки отправки сериализованных данных
private final KafkaTemplate<String, Object> kafkaTemplateForMain;
@Autowired
public JMServiceImpl(JMServiceDao plcPackageDao,
@Qualifier("kafkaTemplateForRetail") KafkaTemplate<String, Object> kafkaTemplateForRetail,
@Qualifier("kafkaTemplateForMain") KafkaTemplate<String, Object> kafkaTemplateForMain) {
this.jmServiceDao = plcPackageDao;
this.kafkaTemplateForRetail = kafkaTemplateForRetail;
this.kafkaTemplateForMain = kafkaTemplateForMain;
}
@Override
public Mono<KafkaSendOKResponse> sendMessage(QueueMessageId queueMessageId) throws IOException {
log.info("Попытка отправить сообщение queueMessageId = {}", queueMessageId.getQueueMessageId());
QueueMessage message = jmServiceDao.getQueueMessage(queueMessageId.getQueueMessageId());
if (message == null) {
throw new ResponseStatusException(NOT_FOUND, "В теле запроса указан несуществующий идентификатор сообщения в очереди");
} else {
KafkaType messagingSystem = KafkaType.valueOf(message.getMessagingSystemCode());
Object messageForSend;
KafkaTemplate<String, Object> kafkaTemplate = switch (messagingSystem) {
case KAFKA_MAIN -> {
ObjectMapper objectMapper = new ObjectMapper();
//строка с данными в JSON
JsonNode rootNode = objectMapper.readTree(message.getMessageData());
//нужна сериализация
SendPTI ptiMessage = new SendPTI();
ptiMessage.setSiebelId(rootNode.get("siebel_id").asText());
ptiMessage.setPti(ByteBuffer.wrap(rootNode.get("pti").asText().getBytes(StandardCharsets.US_ASCII)));
ptiMessage.setPtiDate(Instant.parse(rootNode.get("pti_date").asText()));
messageForSend = ptiMessage;
yield kafkaTemplateForMain;
}
case KAFKA_RETAIL -> {
//отправляем просто строку
messageForSend = message.getMessageData();
yield kafkaTemplateForRetail;
}
default -> throw new IllegalArgumentException("Неподдерживаемая система: " + messagingSystem);
};
log.debug("Отправляем в топик {} данные {}", message.getTopic(), messageForSend);
return Mono.fromFuture(kafkaTemplate.send(message.getTopic(), messageForSend))
.map(result -> {
log.debug("Результат {} отправки сообщения {}", result, queueMessageId.getQueueMessageId());
LocalDateTime localDateTime = result.getRecordMetadata().hasTimestamp()
? LocalDateTime.ofInstant(Instant.ofEpochMilli(result.getRecordMetadata().timestamp()), ZoneId.systemDefault())
: null;
return new KafkaSendOKResponse()
.setQueueMessageId(queueMessageId.getQueueMessageId())
.setSendStatus(1)
.setRecordMetadata(result.getRecordMetadata().toString())
.setSentDate(localDateTime);
})
.doOnError(ex -> {
log.error("Произошла ошибка {} при отправке сообщения {}", ex, queueMessageId.getQueueMessageId());
});
}
}
}
Выборка нужной конфигурации (бина) осуществляется с помощью аннотации @Qualifier. Для KAFKA_MAIN используется Avro сериализатор, для KAFKA_RETAIL отправляется строка.
Для генерации класса SendPTI для сериализации отправляемых\принимаемых данных используется Maven плагин:
<plugin>
<groupId>org.apache.avro</groupId>
<artifactId>avro-maven-plugin</artifactId>
<version>1.11.3</version>
<executions>
<execution>
<phase>generate-sources</phase>
<goals>
<goal>schema</goal>
</goals>
<configuration>
<sourceDirectory>${project.basedir}/src/main/avro/</sourceDirectory>
<outputDirectory>${project.basedir}/src/main/java/</outputDirectory>
</configuration>
</execution>
</executions>
</plugin>В нем нужно указать путь к *.avcs файлу, содержащему исходную схему, мне был предоставлен такой:
{
"type": "record",
"name": "SendPTI",
"namespace": "com.technology.kafkaproducer.jmessageservice.SendPTI",
"fields": [
{
"name": "siebel_id",
"type": "string"
},
{
"name": "pti",
"type": {
"type": "bytes",
"scale": 5,
"precision": 7,
"connect.version": 1,
"connect.parameters": {
"scale": "5",
"connect.decimal.precision": "7"
},
"connect.name": "org.apache.kafka.connect.data.Decimal",
"logicalType": "decimal"
}
},
{
"name": "pti_date",
"type": {
"type": "long",
"connect.version": 1,
"connect.name": "org.apache.kafka.connect.data.Timestamp",
"logicalType": "timestamp-millis"
}
}
],
"connect.name": "ru.COMPANY.schema.SCORF.SendPTI"
}При сборке проекта будет запускаться плагин и будет сгенерирован POJO класс, в моем примере выше это класс SendPTI.
Когда все уже было готово, я столкнулся с другой проблемой; доступ в сеть у меня осуществляется через рабочий прокси, но при попытке отправить сообщение в Kafka, происходила безуспешная попытка подключения к Schema Registry и приложение падало с ошибкой:
2024-05-29 11:09:15.754 [http-nio-8081-exec-2] ERROR i.c.k.s.client.rest.RestService - Failed to send HTTP request to endpoint: https://schema-kafka-test/subjects/SCORF.SendPTI-value/versions/latest/ java.io.IOException: Unable to tunnel through proxy. Proxy returns "HTTP/1.1 407 Proxy Authentication Required"
Решением стало добавление дополнительных параметром при запуске моего локального (разработческого) сервера Томкат:
-Dhttps.proxyHost=proxy.host
-Dhttps.proxyPort=8080
-Dhttps.proxyUser=user
-Dhttps.proxyPassword=superSecretPasswordПосле этого я смог успешно отправлять данные из одного сервиса на два разных сервера Kafka с разными типами авторизации и с сериализацией и без.
Мира!
