Page History

from("kafka:localhost:9092?topic=test&groupId=testing&autoOffsetReset=earliest&consumersCount=1")
						.process(new Processor() {
							@Override
							public void process(Exchange exchange)
									throws Exception {
								String messageKey = "";
								if (exchange.getIn() != null) {
									Message message = exchange.getIn();
									Integer partitionId = (Integer) message
											.getHeader(KafkaConstants.PARTITION);
									String topicName = (String) message
											.getHeader(KafkaConstants.TOPIC);
									if (message.getHeader(KafkaConstants.KEY) != null)
										messageKey = (String) message
												.getHeader(KafkaConstants.KEY);
									Object data = message.getBody();


									System.out.println("topicName :: "
											+ topicName + " partitionId :: "
											+ partitionId + " messageKey :: "
											+ messageKey + " message :: "
											+ data + "\n");
								}
							}
						}).to("log:input");

from("direct:start").process(new Processor() {
					@Override
					public void process(Exchange exchange) throws Exception {
						exchange.getIn().setBody("Test Message from Camel Kafka Component Final",String.class);
						exchange.getIn().setHeader(KafkaConstants.PARTITION_KEY, 0);
						exchange.getIn().setHeader(KafkaConstants.KEY, "1");
					}
				}).to("kafka:localhost:9092?topic=test");