Ignite and Kafka issue

classic Classic list List threaded Threaded
3 messages Options
Alexey Alexey
Reply | Threaded
Open this post in threaded view
|

Ignite and Kafka issue

Hello!
I am trying to execute simple example with Ignite and KafkaConsumer.
Unfortunately I can't get any messages using KafkaConsumer and I can't find where is a problem. Message is always null although consumer started in command line shows all messages that was sent by Producer in my example.
Could you please take a look at my example?

public class KafkaTest2 {
    public static void main(String[] args) throws IgniteException {
        try (Ignite ignite = Ignition.start("examples/config/example-ignite.xml")) {
            Properties prodProps = new Properties();
            prodProps.put("key.serializer",
                    "org.apache.kafka.common.serialization.StringSerializer");
            prodProps.put("value.serializer",
                    "org.apache.kafka.common.serialization.StringSerializer");

            prodProps.put("bootstrap.servers", "localhost:9092");

            try (Producer<String, String> producer = new KafkaProducer<>(prodProps)) {
                ProducerRecord<String, String> data1 = new ProducerRecord<String, String>(
                        "test", "key1", "test-msg1");
                producer.send(data1);
            }

            Properties consumerProps = (Properties) prodProps.clone();
            consumerProps.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            consumerProps.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            consumerProps.put("group.id", "testId");
            consumerProps.put("partition.assignment.strategy", "roundrobin");

            try (Consumer<String, String> consumer = new KafkaConsumer<>(consumerProps)) {
                consumer.subscribe("test");

                Map<String, ConsumerRecords<String, String>> msg  = null;
                while (msg  == null) {
                    msg  = consumer.poll(100);
                }
                System.out.println(msg);
            }
        }
    }
}
Alexey Alexey
Reply | Threaded
Open this post in threaded view
|

Re: Ignite and Kafka issue

The problem was salved by adding dependancies to pom.xml

        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>0.9.0.0</version>
        </dependency>
vkulichenko vkulichenko
Reply | Threaded
Open this post in threaded view
|

Re: Ignite and Kafka issue

Hi Alexey,

The question is actually not about Ignite at all, but I'm glad you solved it :)

Note that Ignite provides KafkaStreamer out of the box, it will automatically consume from Kafka and stream into Ignite. See [1] for details.

[1] https://apacheignite.readme.io/docs/kafka-streamer

-Val