Jason Rosenberg created KAFKA-945:
-------------------------------------
Summary: Problem with test to send a message and then consume it
Key: KAFKA-945
URL: https://issues.apache.org/jira/browse/KAFKA-945
Project: Kafka
Issue Type: Bug
Affects Versions: 0.8
Reporter: Jason Rosenberg
A simple test, which sends on message synchronously, and then consumes it, is
failing, with the latest 0.8 beta release candidate (produced from sha:
23acbd309f5e17de71db46cb6f1a60c8d38ea4e4
Note this problem did not occur with a previous version of the 0.8 branch (e.g.
it seems to work fine for sha: 988d4d8e65a14390abd748318a64e281e4a37c19).
Essentially, it appears that the message never gets sent (after complaining
about missing partition leader, etc.).
To reproduce, run the sample zookeeper and kafka scripts, that come with the
distribution (but first delete all pre-existing state by removing the data
directories that zookeeper and kafka use:
rm -rf /tmp/zookeeper
rm -rf/tmp/kafka_logs
./bin/zookeeper-server-start.sh config/zookeeper.properties
./bin/kafka-server-start.sh config/server.properties
Then execute the simple test code (I will attach a tarball which you should be
able to unpack and run this example).
@Test public void produceAndConsumeMessage() throws Exception {
// assumes zookeeper and kafka server are running.
String zkConnect = "localhost:2181";
int port = 9092;
Properties pProps = new Properties();
pProps.put("metadata.broker.list", "localhost:" + port);
pProps.put("serializer.class", "kafka.serializer.StringEncoder");
ProducerConfig pConfig = new ProducerConfig(pProps);
Producer<Integer, String> producer = new Producer<Integer, String>(pConfig);
KeyedMessage<Integer, String> data =
new KeyedMessage<Integer, String>("test-topic", "test-message");
producer.send(data);
producer.close();
Properties cProps = new Properties();
cProps.put("zookeeper.connect", zkConnect);
cProps.put("group.id", "group1");
ConsumerConfig consumerConfig = new ConsumerConfig(cProps);
ConsumerConnector consumerConnector =
Consumer.createJavaConsumerConnector(consumerConfig);
Map<String, List<KafkaStream<byte[], byte[]>>> topicMessageStreams =
consumerConnector.createMessageStreams(ImmutableMap.of("test-topic",
1));
List<KafkaStream<byte[], byte[]>> streams =
topicMessageStreams.get("test-topic");
final KafkaStream<byte[], byte[]> stream = streams.get(0);
final ConsumerIterator<byte[], byte[]> iter = stream.iterator();
// run in a separate thread
final AtomicBoolean success = new AtomicBoolean(false);
Thread consumeThread = new Thread(new Runnable() {
public void run() {
while (iter.hasNext()) {
byte[] msg = iter.next().message();
String msgStr = new String(msg);
success.set(msgStr.equals("test-message"));
break;
}
}
});
consumeThread.start();
consumeThread.join();
consumerConnector.shutdown();
assertTrue(success.get());
}
--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators
For more information on JIRA, see: http://www.atlassian.com/software/jira
