Java / scala kafka producer не отправляет сообщение в тему

У меня проблема с отправкой сериализованного XML в мою тему Kafka. Всякий раз, когда я запускаю свой код, я не получаю никаких исключений или сообщений об ошибках, но все же я не вижу ни одного из своих сообщений в теме Kafka.

Мои настройки Kafka-Producer:

def WartungsdbKafkaConnector(args: Array[String]): Unit = {  
    val xmlFile = args(0)
    val record = getRecord(xmlFile)
    val kafkaProducer = getKafkaProducer
    kafkaProducer.send(record)
}

protected def getRecord(xmlFile: String): ProducerRecord[String, String] = {
    val lines = scala.io.Source.fromFile(xmlFile).mkString
    val xml = scala.xml.XML.loadString(lines)
    val paramPress = xml \ "PARAMETER" \ "PRESS"
    val databaseId = allCatch.opt {paramPress.\@("NUMBER")}
    val key = databaseId.get
    val topic = args(1)
    new ProducerRecord(topic, key, lines)
}


protected def getKafkaProducer: KafkaProducer[String, String] = {  
    val props = new Properties
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, 
    "ec-x.eu-west-1.compute.amazonaws.com:9092," +
    "ec2-x.eu-west-1.compute.amazonaws.com:9092," +
    "ec2-x.eu-west-1.compute.amazonaws.com:9092")
    props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true")
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer].getName)
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer].getName)
    props.put(ProducerConfig.LINGER_MS_CONFIG, "100")
    props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy")
    props.put(ProducerConfig.RETRIES_CONFIG, "20")
    props.put(ProducerConfig.ACKS_CONFIG, "all")

    new KafkaProducer[String, String](props)
}

Когда я запускаю код, я получаю:

[main] INFO org.apache.kafka.clients.producer.ProducerConfig - ProducerConfig values:
    acks = all
    batch.size = 16384
    bootstrap.servers = [ec2-x.eu-west-1.compute.amazonaws.com:9092, 
    ec2-x.eu-west-1.compute.amazonaws.com:9092, 
    ec2-x.eu-west-1.compute.amazonaws.com:9092]
    buffer.memory = 33554432
    client.id =
    compression.type = snappy
    connections.max.idle.ms = 540000
    enable.idempotence = true
    interceptor.classes = []
    key.serializer = class org.apache.kafka.common.serialization.StringSerializer
    linger.ms = 100
    max.block.ms = 60000
    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 = 32768
    reconnect.backoff.max.ms = 1000
    reconnect.backoff.ms = 50
    request.timeout.ms = 30000
    retries = 2
    retry.backoff.ms = 100
    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
    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 org.apache.kafka.common.serialization.StringSerializer

[main] INFO org.apache.kafka.clients.producer.KafkaProducer - [Producer 
clientId=producer-1] Instantiated an idempotent producer.
[main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka version : 
2.0.0
[main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka commitId : 
3402a8361b734732
[kafka-producer-network-thread | producer-1] INFO 
org.apache.kafka.clients.Metadata - Cluster ID: xeb6oWNpTgSQ_9FHctZ2ng
[kafka-producer-network-thread | producer-1] INFO 
org.apache.kafka.clients.producer.internals.TransactionManager - [Producer 
clientId=producer-1] ProducerId set to 150671 with epoch 0

Любая идея, как заставить его работать? Заранее спасибо!

Вы пробовали получать сообщения темы из командной строки?

Giorgos Myrianthous 10.09.2018 10:34

Пожалуйста, включите свою логику отправки. Код, который вы вставили, создает только производителя, но ничего не отправляет

Mickael Maison 10.09.2018 10:47

@GiorgosMyrianthous: Да из командной строки. Может ли это иметь значение?

ScalaNewbie 10.09.2018 12:02

@MickaelMaison: Отредактировал. Спасибо

ScalaNewbie 10.09.2018 12:02

@ScalaNewbie Тогда можешь поделиться командой?

Giorgos Myrianthous 10.09.2018 12:15

@GiorgosMyrianthous: sudo kafka-console-consumer --bootstrap-server localhost: 9092 --property schema.registry.url = ec2-x.eu-west-1.compute.amazonaws.com:80‌ 81 --topic% topic-name% --from-begin Заранее спасибо :)

ScalaNewbie 10.09.2018 15:16
1
6
905
1
Перейти к ответу Данный вопрос помечен как решенный

Ответы 1

Ответ принят как подходящий

Вы не сбрасываете, не ждете или не закрываете производителя, поэтому приложение просто останавливается, не отправляя данные.

Производители пакетируют данные в течение настраиваемого количества времени и сообщений, чтобы уменьшить количество запросов на отправку, фактически получаемых брокерами.

Пытаться

kafkaProducer.send(record)  // optionally call get() on this to capture the result and potential errors 
kafkaProducer.flush() 
kafkaProducer.close()

Самое главное, никогда не забывайте закрывать производителя (или потребителя)

Другие вопросы по теме