Home | About | Sematext search-lucene.com search-hadoop.com
 Search Hadoop and all its subprojects:

Switch to Threaded View
Kafka, mail # user - Kafka 0.8.0 in maven


Copy link to this message
-
Re: Kafka 0.8.0 in maven
Edward Capriolo 2013-11-07, 03:39
After about 5 days of relentless head pounding, and twittling about
everything under the sun, I figured it out.

If you read to the bottom of this page:
https://cwiki.apache.org/confluence/display/KAFKA/Consumer+Group+Example

You find:

consumerProps.put("auto.offset.reset", "smallest");

Now I can bring up the entire stack in jvm and test like I used to. !
On Wed, Nov 6, 2013 at 2:52 AM, Chris Bedford <[EMAIL PROTECTED]> wrote:

> Hi, Edward..
>
> yup .. you are correct.. when we get to a little over 1000 messages the
> program was  failing with the  exception stack trace i included below.
>
> I fixed the test so it passes as long as the consumer gets all messages
> sent by the producer.. even if an exception is thrown during shut down..
>
> This isn't as clean as i'd like it to be.  I tried
> calling kafkaServer.awaitShutdown();
> And I tried inserting some Thread.sleep() calls to give the consumer and
> producer shut down procedures some time to complete. But I still got the
> stack trace shown below.
>
> I don't have time to chase the bug any further.. But I did correct the
> test, so you can pull it and see that it passes if you want.
>
> Maybe we should file a bug on this...?        It might be that I'm using
> the API incorrectly. I'm not sure at this point.
>
> anyway, thx for informing me of the issue.
>
>
>
>
> *Failure due to broken shut down>> >*
>
> java.io.IOException: Connection reset by peer
> at sun.nio.ch.FileDispatcherImpl.read0(Native Method)
> at sun.nio.ch.SocketDispatcher.read(SocketDispatcher.java:39)
> at sun.nio.ch.IOUtil.readIntoNativeBuffer(IOUtil.java:225)
> at sun.nio.ch.IOUtil.read(IOUtil.java:198)
> at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:375)
> at kafka.utils.Utils$.read(Utils.scala:394)
> at
>
> kafka.network.BoundedByteBufferReceive.readFrom(BoundedByteBufferReceive.scala:54)
> at kafka.network.Processor.read(SocketServer.scala:339)
> at kafka.network.Processor.run(SocketServer.scala:245)
> at java.lang.Thread.run(Thread.java:724)
> INFO 2013-11-05 23:45:53,944
>  ConsumerFetcherThread-RANDOM-GROUP-ID_ubuntu-1383723948022-27b6aba8-0-1
> kafka.consumer.SimpleConsumer Reconnect due to socket error:
> java.nio.channels.ClosedByInterruptException
> at
>
> java.nio.channels.spi.AbstractInterruptibleChannel.end(AbstractInterruptibleChannel.java:202)
> at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:402)
> at sun.nio.ch.SocketAdaptor$SocketInputStream.read(SocketAdaptor.java:220)
> at sun.nio.ch.ChannelInputStream.read(ChannelInputStream.java:103)
> at
> java.nio.channels.Channels$ReadableByteChannelImpl.read(Channels.java:385)
> at kafka.utils.Utils$.read(Utils.scala:394)
> at
>
> kafka.network.BoundedByteBufferReceive.readFrom(BoundedByteBufferReceive.scala:54)
> at kafka.network.Receive$class.readCompletely(Transmission.scala:56)
> at
>
> kafka.network.BoundedByteBufferReceive.readCompletely(BoundedByteBufferReceive.scala:29)
> at kafka.network.BlockingChannel.receive(BlockingChannel.scala:100)
> at kafka.consumer.SimpleConsumer.liftedTree1$1(SimpleConsumer.scala:73)
> at
>
> kafka.consumer.SimpleConsumer.kafka$consumer$SimpleConsumer$$sendRequest(SimpleConsumer.scala:71)
> at
>
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(SimpleConsumer.scala:110)
> at
>
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply(SimpleConsumer.scala:110)
> at
>
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply(SimpleConsumer.scala:110)
> at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
> at
>
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply$mcV$sp(SimpleConsumer.scala:109)
> at
>
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply(SimpleConsumer.scala:109)
> at
>
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply(SimpleConsumer.scala:109)
> at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
> at kafka.consumer.SimpleConsumer.fetch(SimpleConsumer.scala:108)
> at
>
> kafka.server.AbstractFetcherThread.processFetchRequest(AbstractFetcherThread.scala:96)