Error playing with Echo demo: Exception in thread "weave-log-poller" java.lang.IllegalStateException

[Garrett Barton]

Anyone run into this one yet? Trying to patch together the sample code on the github page to see something run.

Exception in thread "weave-log-poller" java.lang.IllegalStateException: Error in fetching: OFFSET_OUT_OF_RANGE
    at com.continuuity.weave.internal.kafka.client.MessageFetcher.received(MessageFetcher.java:74)
    at com.continuuity.weave.internal.kafka.client.KafkaResponseDispatcher.messageReceived(KafkaResponseDispatcher.java:39)
    at org.jboss.netty.channel.SimpleChannelHandler.handleUpstream(SimpleChannelHandler.java:88)
    at org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:558)
    at org.jboss.netty.channel.DefaultChannelPipeline$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:786)
    at org.jboss.netty.channel.Channels.fireMessageReceived(Channels.java:281)
    at com.continuuity.weave.internal.kafka.client.KafkaResponseHandler.messageReceived(KafkaResponseHandler.java:43)
    at org.jboss.netty.channel.SimpleChannelHandler.handleUpstream(SimpleChannelHandler.java:88)
    at org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:558)
    at org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:553)
    at org.jboss.netty.channel.Channels.fireMessageReceived(Channels.java:268)
    at org.jboss.netty.channel.Channels.fireMessageReceived(Channels.java:255)
    at org.jboss.netty.channel.socket.nio.NioWorker.read(NioWorker.java:84)
    at org.jboss.netty.channel.socket.nio.AbstractNioWorker.processSelectedKeys(AbstractNioWorker.java:471)
    at org.jboss.netty.channel.socket.nio.AbstractNioWorker.run(AbstractNioWorker.java:332)
    at org.jboss.netty.channel.socket.nio.NioWorker.run(NioWorker.java:35)
    at org.jboss.netty.util.ThreadRenamingRunnable.run(ThreadRenamingRunnable.java:102)
    at org.jboss.netty.util.internal.DeadLockProofWorker$1.run(DeadLockProofWorker.java:42)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1146)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
    at java.lang.Thread.run(Thread.java:679)

Got this in my stderr file inside yarn:

13/07/01 18:20:31 INFO server.KafkaServer: Starting Kafka server...
13/07/01 18:20:31 INFO log.LogManager: No log directory found, creating '/tmp/hadoop-gman/nm-local-dir/usercache/gman/appcache/application_1372716741232_0001/container_1372716741232_0001_01_000001/kafka-logs'
13/07/01 18:20:31 INFO log.LogManager: starting log cleaner every 60000 ms
13/07/01 18:20:31 INFO server.KafkaZooKeeper: connecting to ZK: localhost:2181/App/e67b9dca-f9a4-4b9e-9747-1cd4c93db3f8/kafka
13/07/01 18:20:31 INFO zkclient.ZkEventThread: Starting ZkClient event thread.
13/07/01 18:20:31 INFO zkclient.ZkClient: zookeeper state changed (SyncConnected)
13/07/01 18:20:31 INFO network.Acceptor: Awaiting connections on port 38860
13/07/01 18:20:31 INFO utils.Mx4jLoader$: Will not load MX4J, mx4j-tools.jar is not in the classpath
13/07/01 18:20:31 INFO server.KafkaZooKeeper: Registering broker /brokers/ids/0
13/07/01 18:20:31 INFO server.KafkaZooKeeper: Registering broker /brokers/ids/0 succeeded with id:0,creatorId:127.0.1.1-1372717231523,host:127.0.1.1,port:38860
13/07/01 18:20:31 INFO log.LogManager: Starting log flusher every 1000 ms with the following overrides Map()
13/07/01 18:20:31 INFO server.KafkaServer: Kafka server started.
13/07/01 18:20:31 INFO log.LogManager: Created log for 'log'-0
13/07/01 18:20:31 INFO server.KafkaZooKeeper: Begin registering broker topic /brokers/topics/log/0 with 1 partitions
13/07/01 18:20:31 INFO server.KafkaZooKeeper: End registering broker topic /brokers/topics/log/0
13/07/01 18:20:34 ERROR server.KafkaRequestHandlers: error when processing request FetchRequest(topic:log, part:0 offset:5025 maxSize:1048576)
kafka.common.OffsetOutOfRangeException: offset 5025 is out of range
	at kafka.log.Log$.findRange(Log.scala:46)
	at kafka.log.Log.read(Log.scala:264)
	at kafka.server.KafkaRequestHandlers.kafka$server$KafkaRequestHandlers$$readMessageSet(KafkaRequestHandlers.scala:112)
	at kafka.server.KafkaRequestHandlers.handleFetchRequest(KafkaRequestHandlers.scala:92)
	at kafka.server.KafkaRequestHandlers$$anonfun$handlerFor$2.apply(KafkaRequestHandlers.scala:39)
	at kafka.server.KafkaRequestHandlers$$anonfun$handlerFor$2.apply(KafkaRequestHandlers.scala:39)
	at kafka.network.Processor.handle(SocketServer.scala:296)
	at kafka.network.Processor.read(SocketServer.scala:319)
	at kafka.network.Processor.run(SocketServer.scala:214)
	at java.lang.Thread.run(Thread.java:722)
13/07/01 18:21:05 INFO network.Processor: Closing socket connection to /127.0.0.1.

[Terence Yim]

Hi Garrett,

Can you give me some more details? How do you run the echo demo? Using mvn test or running against your own cluster? If it's the latter case, which version of hadoop are you using?

Also, besides the error you see, does the EchoServer actually running and accepting requests?

Thanks,

Terence

[Garrett Barton]

Hey Terence,
This is vanilla apache 2.0.5-alpha that I am playing with.  I started with mostly pasting your example code from the tutorial over into a class to try out.  I just found your actual test and I can get it to run fine against my Yarn instance.  I think there is something subtle about initialization that I do not understand. With this controller setup:

        WeaveController controller = runnerService
                .prepare(
                        new App(),
                        ResourceSpecification.Builder.with().setCores(1)
                                .setMemory(1, ResourceSpecification.SizeUnit.GIGA).setInstances(2).build())
                .addLogHandler(new PrinterLogHandler(new PrintWriter(System.out, true)))
                .withApplicationArguments("echo").withArguments("App", "echo2").start();

This initialize will NOT work, causes my error:

    @Override
    public void initialize(WeaveContext context) {
        super.initialize(context);
        running = true;
        LOG.info("initialize method called");

        try {
            serverSocket = new ServerSocket(0);
            LOG.info("App came up on port " + serverSocket.getLocalPort());
            context.announce("echo", serverSocket.getLocalPort());
        } catch (IOException e) {
            LOG.error("well damn, init filed", e);
        }
    }

Yet this one works fine (copied from your EchoServer):
    @Override
    public void initialize(WeaveContext context) {
        super.initialize(context);
        running = true;
        try {
            serverSocket = new ServerSocket(0);
            LOG.info("EchoServer started: " + serverSocket.getLocalSocketAddress() + ", id: " + context.getInstanceId()
                    + ", count: " + context.getInstanceCount());
            context.announce(context.getApplicationArguments()[0], serverSocket.getLocalPort());
            context.announce(context.getArguments()[0], serverSocket.getLocalPort());
        } catch (IOException e) {
            throw Throwables.propagate(e);
        }
    }

Is there something significant about grabbing the context that I dont know?

Also is there a way to lower the default debug level, I get so much dfs output at the moment its kinda hard to debug.

Thanks for the reply!
~Garrett

[Terence Yim]

This ia quite strange. I don't see anything thing wrong from your initialize method. Can you paste me the full source of your version of EchoServer and also the client class (the one do prepare and start) so that I could reproduce the problem and debug it.

Terence

[Garrett Barton]

I've attached the whole class, condensed it into a single one so its easier to run. Hopefully something stands out for ya!

~Garrett