Spring Cloud Stream Kafka с включенной транзакцией и динамическим пунктом назначения.У меня есть две разные службы
- Первая служба будет прослушивать очередь Solace и выводить ее в kafka topic-1 (где транзакции включены)
- Вторая служба будет прослушивать сверху kafka topic-1 и записать его в другую тему kafka-2 (где у нас нет ручных фиксаций, транзакций, разрешенных для создания в другой теме, автоматического смещения фиксации как false, а для изоляционного элемента установлено значение read_commited), и мы будем динамически идентифицировать имена тем, чтобы мы моглииспользование динамического распознавателя назначения
Теперь у меня проблема со вторым сервисом, если я просто запускаю сервис как @StreamListener & @SendTo, он работает нормально, как и ожидалось.Но когда я начал использовать динамическое назначение, у меня возникает следующая проблема:
динамическое назначение
Caused by: java.lang.IllegalStateException: Cannot perform operation after producer has been closed
at org.apache.kafka.clients.producer.KafkaProducer.throwIfProducerClosed(KafkaProducer.java:810) ~[kafka-clients-2.0.0.jar:na]
at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:819) ~[kafka-clients-2.0.0.jar:na]
at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:803) ~[kafka-clients-2.0.0.jar:na]
at org.springframework.kafka.core.DefaultKafkaProducerFactory$CloseSafeProducer.send(DefaultKafkaProducerFactory.java:423) ~[spring-kafka-2.2.0.RELEASE.jar:2.2.0.RELEASE]
at org.springframework.kafka.core.KafkaTemplate.doSend(KafkaTemplate.java:351) ~[spring-kafka-2.2.0.RELEASE.jar:2.2.0.RELEASE]
at org.springframework.kafka.core.KafkaTemplate.send(KafkaTemplate.java:209) ~[spring-kafka-2.2.0.RELEASE.jar:2.2.0.RELEASE]
at org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler.handleRequestMessage(KafkaProducerMessageHandler.java:382) ~[spring-integration-kafka-3.1.0.RELEASE.jar:3.1.0.RELEASE]
at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:123) [spring-integration-core-5.1.0.RELEASE.jar:5.1.0.RELEASE]
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:169) [spring-integration-core-5.1.0.RELEASE.jar:5.1.0.RELEASE]
пробовал оба подхода для динамического распознавания назначения: динамическое определение назначения
yml для весеннего облака kafka:
spring:
cloud.stream:
bindings:
input:
destination: test_input
content-type: application/json
group: test_group
output:
destination: test_output
content-type: application/json
kafka.binder:
configuration:
isolation.level: read_committed
security.protocol: SASL_SSL
sasl.mechanism: GSSAPI
sasl.kerberos.service.name: kafka
ssl.truststore.location: jks
ssl.truststore.password:
ssl.endpoint.identification.algorithm: null
brokers: broker1:9092,broker2:9092,broker3:9092
auto-create-topics: false
transaction:
transaction-id-prefix: trans-2
producer:
configuration:
retries: 2000
acks: all
security.protocol: SASL_SSL
sasl.mechanism: GSSAPI
sasl.kerberos.service.name: kafka
ssl.truststore.location: jks
ssl.truststore.password:
ssl.endpoint.identification.algorithm: null
Вот фон этого вопроса
Spring Cloud Stream для Kafka с API потребителя / производителя ровно после того, как семантика с префиксом-ID-транзакции не работает, так какожидается
Обновлен код:
@Autowired
private BinderAwareChannelResolver resolver;
@StreamListener(target = Processor.INPUT)
public void consumer(@Payload Object inMessage, @Headers Map headers) {
String topicName = null;
String itemType = null;
try {
TransactionSynchronizationManager.setActualTransactionActive(true);
itemType = msgService.itemTypeExtract((String) inMessage);
topicName = msgService.getTopicName(itemType, (String) inMessage);
Map<String, Object> headersMap = new HashMap<>();
headersMap.put(MessageHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE);
resolver.resolveDestination("destination_topic")
.send(MessageBuilder.createMessage(inMessage, new MessageHeaders(headersMap)), 10000);
} catch (Exception e) {
LOGGER.error("error " + e.getMessage());
}
}