Using Message and Message Set like KeyedMessage

[mesarvagya]

On the official kafka page https://cwiki.apache.org/confluence/display/KAFKA/0.8.0+Producer+Example , consider

    Properties props = new Properties();
    props.put("metadata.broker.list", "broker1:9092,broker2:9092 ");
    props.put("serializer.class", "kafka.serializer.StringEncoder");
    props.put("partitioner.class", "example.producer.SimplePartitioner");
    props.put("request.required.acks", "1");

    ProducerConfig config = new ProducerConfig(props);

    Producer<String, String> producer = new Producer<String, String>(config);

    for (long nEvents = 0; nEvents < events; nEvents++) { 
           long runtime = new Date().getTime();  
           String ip = “192.168.2.” + rnd.nextInt(255); 
           String msg = runtime + “,www.example.com,” + ip; 
           KeyedMessage<String, String> data = new KeyedMessage<String, String>("page_visits", ip, msg);
           producer.send(data);
    }

I tried to emulate this piece of code into its equivalent C++ code. Based on your example code. I have tried:

 Client::Configuration configuration;
 configuration.auto_connect = true;
 configuration.client_id = "libkafka_asio_example";
 configuration.socket_timeout = 10000;
 configuration.AddBrokerFromString("192.168.2.60:9092");
  boost::asio::io_service ios;
  Client client(ios, configuration);

  ProduceRequest request;

  int i(0);
  while(i < 100)
  { 
     std::string msg = "Hello world " + i;
  libkafka_asio::Message message;
  // How to construct the message like KeyedMessage here and send it to Kafka.
  client.AsyncRequest(request, &HandleRequest);
  ios.run();
  delete message;
  ++i;
   }

I am using Windows VS2010.

[danieljoos]

Hi,

The library currently only implements the basic protocol of Kafka. The Producer class in Kafka does quite some work behind the scenes. It needs to get some metadata for the specified topic (in your case 'page_visits') and it uses some partitioner to calculate the partition, to which it wants to produce the message. Implementing this with libkafka-asio could look like this: https://gist.github.com/danieljoos/59c6867f7d316505dad3

In line 138 ff., the example shows how the partition number can be calculated using the FNV-1a hashing algorithm. This is basically what Kafka's KeyedMessage in combination with SimplePartitioner does (please correct me if I'm wrong).

Tested it with VS 2010.

[mesarvagya]

Hi danieljoos. Thanks for the insight on how the emulate the code. I will give it a go and notify you if it works as expected. Thank again.

[mesarvagya]

Hi Daniel, I ran your code and viewed the output in windows using kafka-console-consumer.bat. It did produced the message. Thanks once again.