java - kafkatemplate - spring kafka bootstrap servers

Retraso en los consumidores que consumen mensajes en Apache Kafka

Question

Estoy usando Kafka 0.8.0 y tratando de lograr el escenario mencionado a continuación.

API de JCA (actúa como productor y envía datos a) -----> Consumer ------> HBase

Estoy enviando cada mensaje al consumidor tan pronto como obtengo los datos utilizando el Cliente JCA. Por ejemplo, tan pronto como el productor envíe el mensaje n. ° 1, quiero obtener lo mismo del consumidor y ''poner'' en HBase. Pero mi consumidor comienza a buscar los mensajes después de algunos n mensajes aleatorios. Quiero sincronizar al productor y al consumidor para que ambos comiencen a trabajar juntos.

He utilizado:

1 corredor

1 tema único

1 productor único y consumidor de alto nivel.

¿Alguien puede sugerir qué debo hacer para lograr lo mismo?

EDITADO:

Añadiendo un fragmento de código relevante.

Consumer.java

public class Consumer extends Thread {
    private final ConsumerConnector consumer;
    private final String topic;
    PrintWriter pw = null;
    int t = 0;
    StringDecoder kd = new StringDecoder(null);
    Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
    Map<String, List<KafkaStream<String, Signal>>> consumerMap;
    KafkaStream<String, Signal> stream;
    ConsumerIterator<String, Signal> it;

    public Consumer(String topic) {
        consumer = kafka.consumer.Consumer.createJavaConsumerConnector(createConsumerConfig());

        this.topic = topic;
        topicCountMap.put(topic, new Integer(1));
        consumerMap = consumer.createMessageStreams(topicCountMap, kd, new Serializer(
                new VerifiableProperties()));
        stream = consumerMap.get(topic).get(0);
        it = stream.iterator();

    }

    private static ConsumerConfig createConsumerConfig() {
        Properties props = new Properties();
        props.put("zookeeper.connect", KafkaProperties.zkConnect);
        props.put("group.id", KafkaProperties.groupId);
        props.put("zookeeper.session.timeout.ms", "400");
        props.put("zookeeper.sync.time.ms", "200");
        props.put("auto.commit.interval.ms", "1000");
        props.put("fetch.size", "1024");

        return new ConsumerConfig(props);

    }

    synchronized public void run() {

        while (it.hasNext()) {
            t = (it.next().message()).getChannelid();
            System.out.println("In Consumer received msg" + t);
        }
    }
}

productor.java

public class Producer {
    public final kafka.javaapi.producer.Producer<String, Signal> producer;
    private final String topic;
    private final Properties props = new Properties();

    public Producer(String topic)
    {
        props.put("serializer.class", "org.bigdata.kafka.Serializer");
        props.put("key.serializer.class", "kafka.serializer.StringEncoder");
        props.put("metadata.broker.list", "localhost:9092");
        // Use random partitioner. Don''t need the key type. Just set it to Integer.
        // The message is of type userdefined Object .
        producer = new kafka.javaapi.producer.Producer<String,Signal(newProducerConfig(props));
        this.topic = topic;
    }
}

KafkaProperties.java

public interface KafkaProperties {
    final static String zkConnect = "127.0.0.1:2181";
    final static String groupId = "group1";
    final static String topic = "test00";
    final static String kafkaServerURL = "localhost";
    final static int kafkaServerPort = 9092;
    final static int kafkaProducerBufferSize = 64 * 1024;
    final static int connectionTimeOut = 100000;
    final static int reconnectInterval = 10000;
    final static String clientId = "SimpleConsumerDemoClient";
}

Así es como el consumidor se está comportando para los primeros 10 mensajes que no envía el mensaje recibido por el consumidor, pero a partir del mensaje 11 comienza a funcionar correctamente.

     producer sending msg1

     producer sending msg2

     producer sending msg3

     producer sending msg4

     producer sending msg5

     producer sending msg6

     producer sending msg7

     producer sending msg8

     producer sending msg9

     producer sending msg10

     producer sending msg11

     producer sending msg12
     In Consumer received msg12

     producer sending msg13
     In Consumer received msg13

     producer sending msg14
     In Consumer received msg14

     producer sending msg15
     In Consumer received msg15

     producer sending msg16
     In Consumer received msg16

     producer sending msg17
     In Consumer received msg17

     producer sending msg18
     In Consumer received msg18

     producer sending msg19
     In Consumer received msg19

     producer sending msg20
     In Consumer received msg20

     producer sending msg21
     In Consumer received msg21

EDITADO: agregando la función de escucha donde el productor está enviando mensajes al consumidor. Y estoy usando la configuración del productor por defecto no la sobrescribí

public synchronized void onValueChanged(final MonitorEvent event_) {


    // Get the value from the DBR
    try {
        final DBR dbr = event_.getDBR();

        final String[] val = (String[]) dbr.getValue();

        producer1.producer.send(new KeyedMessage<String, Signal>         
                    (KafkaProperties.topic,new Signal(messageNo)));
        System.out.println("producer sending msg"+messageNo);

        messageNo++;


    } catch (Exception ex) {
        ex.printStackTrace();
    }
}

Answer 1

1. Intente agregar props.put("request.required.acks", "1") a la configuración del productor. Por defecto, el productor no espera recibos y la entrega de mensajes no está garantizada. Por lo tanto, si inicia Broker justo antes de su prueba, el productor puede comenzar a enviar mensajes antes de que el broker se inicie por completo y se pierdan los primeros mensajes.

2. Intente agregar props.put("auto.offset.reset", "smallest") a la configuración del consumidor. Es igual a la opción --from --from-beginning begin de kafka-console-consumer.sh. Si su consumidor comienza más tarde que el productor y no hay datos de compensación guardados en Zookeeper, entonces, de forma predeterminada, comenzará a consumir solo mensajes nuevos (consulte la sección Config. Del consumidor en documentos).

Answer 2

Esto puede deberse a más no de particiones que no de consumidores. Verifique si el tema se crea solo con una sola partición, entonces no perderá ningún mensaje en el consumidor.