[
https://issues.apache.org/jira/browse/KAFKA-1632?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14133345#comment-14133345
]
aarti gupta commented on KAFKA-1632:
------------------------------------
Agreed, so to process a single message, one would call head() on the stream?
which would block till a message is available for consumption, returning the
message when it returns?
Am i missing some 'Runtime' dependency? (that could explain the runtime
exception thrown on calling head()?)
> No such method error on KafkaStream.head
> ----------------------------------------
>
> Key: KAFKA-1632
> URL: https://issues.apache.org/jira/browse/KAFKA-1632
> Project: Kafka
> Issue Type: Bug
> Affects Versions: 0.8.0
> Reporter: aarti gupta
>
> \The following error is thrown, (when I call KafkaStream.head(), as shown in
> the code snippet below)
> WARN - java.lang.NoSuchMethodError:
> kafka.consumer.KafkaStream.head()Lkafka/message/MessageAndMetadata;
> My use case, is that I want to block on the receive() method, and when a
> message is published on the topic, I 'return the head' of the queue to the
> calling method, that processes it.
> I do not use partitioning and have a single stream.
> import com.google.common.collect.Maps;
> import x.x.x.Task;
> import kafka.consumer.ConsumerConfig;
> import kafka.consumer.KafkaStream;
> import kafka.javaapi.consumer.ConsumerConnector;
> import kafka.javaapi.consumer.ZookeeperConsumerConnector;
> import kafka.message.MessageAndMetadata;
> import org.codehaus.jettison.json.JSONException;
> import org.codehaus.jettison.json.JSONObject;
> import org.slf4j.Logger;
> import org.slf4j.LoggerFactory;
> import java.io.IOException;
> import java.util.List;
> import java.util.Map;
> import java.util.Properties;
> import java.util.concurrent.Callable;
> import java.util.concurrent.Executors;
> /**
> * @author agupta
> */
> public class KafkaConsumerDelegate implements ConsumerDelegate {
> private ConsumerConnector consumerConnector;
> private String topicName;
> private static Logger LOG =
> LoggerFactory.getLogger(KafkaProducerDelegate.class.getName());
> private final Map<String, Integer> topicCount = Maps.newHashMap();
> private Map<String, List<KafkaStream<byte[], byte[]>>> messageStreams;
> private List<KafkaStream<byte[], byte[]>> kafkaStreams;
> @Override
> public Task receive(final boolean consumerConfirms) {
> try {
> LOG.info("Kafka consumer delegate listening on topic " +
> getTopicName());
> kafkaStreams = messageStreams.get(getTopicName());
> final KafkaStream<byte[], byte[]> kafkaStream =
> kafkaStreams.get(0);
> return Executors.newSingleThreadExecutor().submit(new
> Callable<Task>() {
> @Override
> public Task call() throws Exception {
> final MessageAndMetadata<byte[], byte[]>
> messageAndMetadata= kafkaStream.head();
> final Task message = new Task() {
> @Override
> public byte[] getBytes() {
> return messageAndMetadata.message();
> }
> };
> return message;
> }
> }).get();
> } catch (Exception e) {
> LOG.warn("Error in consumer " + e.getMessage());
> }
> return null;
> }
> @Override
> public void initialize(JSONObject configData, boolean publisherAckMode)
> throws IOException {
> try {
> this.topicName = configData.getString("topicName");
> LOG.info("Topic name is " + topicName);
> } catch (JSONException e) {
> e.printStackTrace();
> LOG.error("Error parsing configuration", e);
> }
> Properties properties = new Properties();
> properties.put("zookeeper.connect", "localhost:2181");
> properties.put("group.id", "testgroup");
> ConsumerConfig consumerConfig = new ConsumerConfig(properties);
> //only one stream, and one topic, (Since we are not supporting
> partitioning)
> topicCount.put(getTopicName(), 1);
> consumerConnector = new ZookeeperConsumerConnector(consumerConfig);
> messageStreams = consumerConnector.createMessageStreams(topicCount);
> }
> @Override
> public void stop() throws IOException {
> //TODO
> throw new UnsupportedOperationException("Method Not Implemented");
> }
> public String getTopicName() {
> return this.topicName;
> }
> }
> Lastly, I am using the following binary
> kafka_2.8.0-0.8.1.1
> and the following maven dependency
> <dependency>
> <groupId>org.apache.kafka</groupId>
> <artifactId>kafka_2.10</artifactId>
> <version>0.8.1.1</version>
> </dependency>
> Any suggestions?
> Thanks
> aarti
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
