Camel kafka отправляет сообщение до установления соединения - PullRequest
0 голосов
/ 13 марта 2020

У меня есть маршрут Артемида -> Кафка:

from("artemis:messageQueue")
      .routeId("myRoute")
      .to("kafka:myTopic?serializerClass=mySerializer&clientId=myClientId");

Все работает нормально, но при запуске приложения Spring Camel пытается отправить сообщение до того, как соединение установлено.

Я ищет решение, как заставить Camel прекратить отправку сообщения до установления соединения. Я не нашел ничего об этом.

И у меня есть ошибка:

2020-03-11T18:59:12,221|ERROR|Camel (camel-1) thread #5 - JmsConsumer[messageQueue]|o.a.c.p.DefaultErrorHandler|Failed delivery for (MessageId: ID:3cef05a7-63b1-11ea-bf18-0242fcc61e15 on ExchangeId: ID-BO-TST-CS10-1583942309253-0-2). Exhausted after delivery attempt: 1 caught: org.apache.kafka.common.errors.TimeoutException: Topic myTopic not present in metadata after 5000 ms.

Message History
---------------------------------------------------------------------------------------------------------------------------------------
RouteId              ProcessorId          Processor                                                                        Elapsed (ms)
[myRoute  ] [myRoute  ] [artemis://queue:messageQueue                                              ] [      5084]
[myRoute  ] [to38              ] [kafka:myTopic?serializerClass=mySerializer] [      5068]

Stacktrace
---------------------------------------------------------------------------------------------------------------------------------------
org.apache.kafka.common.errors.TimeoutException: Topic myTopic not present in metadata after 5000 ms.

Когда соединение установлено, все работает нормально.

2020-03-11T18:59:37,105|INFO|kafka-producer-network-thread|o.a.k.c.Metadata|[Producer clientId=myClientId] Cluster ID: lG6hlUa1RhiYhbqjEBxK6g

Производитель Kafka Клиентские конфигурации:

2020-03-11T18:59:03,755|INFO|main|o.a.k.c.p.ProducerConfig|ProducerConfig values: 
    acks = all
    batch.size = 16384
    bootstrap.servers = [myKafkaBroker:9092]
    buffer.memory = 33554432
    client.dns.lookup = default
    client.id = myClientId
    compression.type = none
    connections.max.idle.ms = 540000
    delivery.timeout.ms = 120000
    enable.idempotence = false
    interceptor.classes = []
    key.serializer = class org.apache.kafka.common.serialization.StringSerializer
    linger.ms = 0
    max.block.ms = 5000
    max.in.flight.requests.per.connection = 5
    max.request.size = 1048576
    metadata.max.age.ms = 300000
    metric.reporters = []
    metrics.num.samples = 2
    metrics.recording.level = INFO
    metrics.sample.window.ms = 30000
    partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner
    receive.buffer.bytes = 65536
    reconnect.backoff.max.ms = 1000
    reconnect.backoff.ms = 30000
    request.timeout.ms = 2000
    retries = 0
    retry.backoff.ms = 1000
    sasl.client.callback.handler.class = null
    sasl.jaas.config = null
    sasl.kerberos.kinit.cmd = /usr/bin/kinit
    sasl.kerberos.min.time.before.relogin = 60000
    sasl.kerberos.service.name = null
    sasl.kerberos.ticket.renew.jitter = 0.05
    sasl.kerberos.ticket.renew.window.factor = 0.8
    sasl.login.callback.handler.class = null
    sasl.login.class = null
    sasl.login.refresh.buffer.seconds = 300
    sasl.login.refresh.min.period.seconds = 60
    sasl.login.refresh.window.factor = 0.8
    sasl.login.refresh.window.jitter = 0.05
    sasl.mechanism = GSSAPI
    security.protocol = PLAINTEXT
    security.providers = null
    send.buffer.bytes = 131072
    ssl.cipher.suites = null
    ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
    ssl.endpoint.identification.algorithm = https
    ssl.key.password = null
    ssl.keymanager.algorithm = SunX509
    ssl.keystore.location = null
    ssl.keystore.password = null
    ssl.keystore.type = JKS
    ssl.protocol = TLS
    ssl.provider = null
    ssl.secure.random.implementation = null
    ssl.trustmanager.algorithm = PKIX
    ssl.truststore.location = null
    ssl.truststore.password = null
    ssl.truststore.type = JKS
    transaction.timeout.ms = 60000
    transactional.id = null
    value.serializer = class mySerializer

Я использую Camel 2.24.2, Camel-Kafka-Starter для Spring Boot, Кафка-клиенты: 2.4.0

...