Apache Kafka не потребляет от API - PullRequest
Новая
       7

Apache Kafka не потребляет от API

0 голосов
/ 25 мая 2018

Консольные команды kafka-console-producer.sh и kafka-console-consumer.sh работают правильно, но когда я пытаюсь производить или потреблять с помощью API, я не могу!Может кто-нибудь сказать мне, если что-то не так с моим кодом Scala? 

import java.util.Properties

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}

object ScalaProducerExample  {
  val topic = "test"
  val brokers = "<broker>:9092"
  val props = new Properties()
  props.put("bootstrap.servers", brokers)
  props.put("client.id", "ScalaProducerExample")
  props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
  props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
  val producer = new KafkaProducer[String, String](props)
  val data = new ProducerRecord[String, String](topic, "message")
  producer.send(data)
  producer.close()
}

Это зависимости, загруженные в файл build.sbt: 

libraryDependencies += "org.apache.kafka" % "kafka-clients" % "0.8.2.1"

libraryDependencies += "org.apache.kafka" %% "kafka" % "0.10.2.0"

Я даже написал это на Java ито же самое происходит

import org.apache.kafka.clients.ClientRequest;
import org.apache.kafka.clients.ClientResponse;
import org.apache.kafka.clients.KafkaClient;
import org.apache.kafka.clients.RequestCompletionHandler;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.requests.AbstractRequest;

import java.io.IOException;
import java.util.Date;
import java.util.List;
import java.util.Properties;
import java.util.Random;

public class ProducerExample {
    public static void main(String[] args) {
        String topic = "test";
        String brokers = "<broker>:9092";
        System.out.println("init " );
        Properties props = new Properties();
        props.put("bootstrap.servers", brokers);
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);


        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        System.out.println("creating prducer " );
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);
        producer.flush();
        producer.send(new ProducerRecord<>(topic, "1", "2"));
        producer.close();
        System.out.println("close  " );
    }
}

Зависимости в build.sbt: 

libraryDependencies += "org.apache.kafka" % "kafka-clients" % "0.8.2.1"

Я знаю, что соединение работает, потому что, когда я меняю брокера, я получаю сообщение об ошибке.Но когда брокер прав, программа успешно работает, но я не получаю никаких сообщений.

Обновление: Я предполагаю, что причина, по которой программа работает успешно, заключается в том, что она дает тайм-аут.Я запустил это 

try {
            producer.send(new ProducerRecord<>(topic, "1", "2")).get(30, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }

И получил эту ошибку: 

java.util.concurrent.TimeoutException: Timeout after waiting for 30000 ms.
        at org.apache.kafka.clients.producer.internals.FutureRecordMetadata.get(FutureRecordMetadata.java:64)
        at org.apache.kafka.clients.producer.internals.FutureRecordMetadata.get(FutureRecordMetadata.java:25)
        at de.innocow.kafka.ProducerExample.main(ProducerExample.java:45)

Как мне отладить больше, чем это, и выяснить, почему производитель не отправляет?

1 Ответ

0 голосов
/ 25 мая 2018 
producer.send(new ProducerRecord<>(topic, "1", "2"));
producer.flush();            
producer.close();

Попробуйте с этим и посмотрите Документы : 

The flush() call gives a convenient way to ensure all previously sent messages have actually completed. 
 This example shows how to consume from one Kafka topic and produce to another Kafka topic:

 for(ConsumerRecord<String, String> record: consumer.poll(100))
     producer.send(new ProducerRecord("my-topic", record.key(), record.value());
 producer.flush();
 consumer.commit();
...