Apache Storm java.nio.channels.ClosedChannelException: null

Question

We're trying to use Apache Storm for processing large amount of (fake) messages. Message example:

"{"clientName":"Sergey Bakulin","sum":12925,"group":"property","suspicious":false,"clientId":2,"dt":1404387303764,"coord":{"lat":55.767842588357645,"lon":37.46920361823332}}".

We're using Apache Kafka as a source of messages for our Storm cluster. Our purpose is to be able to process at least 50k msg/sec/node. In case when we use more than one node we constantly stuck with the error (log snippet is from worker-*.log):

2014-07-03 15:14:47 b.s.m.n.Client [INFO] failed to send requests to ip-172-31-23-123.eu-west-1.compute.internal/172.31.23.123:6701: java.nio.channels.ClosedChannelException: null
at org.jboss.netty.channel.socket.nio.AbstractNioWorker.cleanUpWriteBuffer(AbstractNioWorker.java:381) [netty-3.6.3.Final.jar:na]
at org.jboss.netty.channel.socket.nio.AbstractNioWorker.close(AbstractNioWorker.java:349) [netty-3.6.3.Final.jar:na]
at org.jboss.netty.channel.socket.nio.NioWorker.read(NioWorker.java:93) [netty-3.6.3.Final.jar:na]
at org.jboss.netty.channel.socket.nio.AbstractNioWorker.process(AbstractNioWorker.java:107) [netty-3.6.3.Final.jar:na]
at org.jboss.netty.channel.socket.nio.AbstractNioSelector.run(AbstractNioSelector.java:312) [netty-3.6.3.Final.jar:na]
at org.jboss.netty.channel.socket.nio.AbstractNioWorker.run(AbstractNioWorker.java:88) [netty-3.6.3.Final.jar:na]
at org.jboss.netty.channel.socket.nio.NioWorker.run(NioWorker.java:178) [netty-3.6.3.Final.jar:na]
at org.jboss.netty.util.ThreadRenamingRunnable.run(ThreadRenamingRunnable.java:108) [netty-3.6.3.Final.jar:na]
at org.jboss.netty.util.internal.DeadLockProofWorker$1.run(DeadLockProofWorker.java:42) [netty-3.6.3.Final.jar:na]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) [na:1.7.0_51]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) [na:1.7.0_51]
at java.lang.Thread.run(Thread.java:744) [na:1.7.0_51]
2014-07-03 15:14:47 b.s.m.n.StormClientErrorHandler [INFO] Connection failed Netty-Client-ip-172-31-23-123.eu-west-1.compute.internal/172.31.23.123:6701

Our current storm config:

########### These MUST be filled in for a storm configuration
storm.zookeeper.servers:
  - "172.31.*.*"

storm.local.dir: "/home/*/storm/data"
nimbus.host: "127.0.0.1"
supervisor.slots.ports:
  - 6701
  - 6702

ui.port: 8090

worker.childopts: "-Xmx6g -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.port=1%ID% -Dcom.sun.management.jmxremote.local.only=false -Dcom.sun$

supervisor.childopts: "-Xmx1024m -Djava.net.preferIPv4Stack=true"
supervisor.worker.start.timeout.secs: 10
supervisor.worker.timeout.secs: 10
supervisor.monitor.frequency.secs: 3
supervisor.heartbeat.frequency.secs: 5
supervisor.enable: true

storm.messaging.netty.server_worker_threads: 2
storm.messaging.netty.client_worker_threads: 2
storm.messaging.netty.buffer_size: 5242880
storm.messaging.netty.max_retries: 25
storm.messaging.netty.max_wait_ms: 1000

Our storm topology:

Properties conf = Util.readProperties(ClientTopology.class, "storm.properties");

prepareRedisDB(conf);

TopologyBuilder builder = new TopologyBuilder();

builder.setSpout("kafka_trans_spout", getKafkaSpout(conf, conf.getProperty("kafka_trans_topic")), 3);
builder.setSpout("kafka_socevent_spout", getKafkaSpout(conf, conf.getProperty("kafka_socevent_topic")), 3);

builder.setBolt("json_to_tuple_trans_bolt", new JSONToTupleBolt(Transaction.class), 6)
        .shuffleGrouping("kafka_trans_spout");
builder.setBolt("json_to_tuple_socevent_bolt", new JSONToTupleBolt(SocialEvent.class), 3)
        .shuffleGrouping("kafka_socevent_spout");

builder.setBolt("alert_bolt", new AlertBolt(conf), 3)
        .fieldsGrouping("json_to_tuple_trans_bolt", new Fields("cl_id"))
        .fieldsGrouping("json_to_tuple_socevent_bolt", new Fields("cl_id"));
builder.setBolt("offer_bolt", new NearestOfferBolt(conf), 3)
        .shuffleGrouping("json_to_tuple_trans_bolt");

run(builder, args, 6);

private static KafkaSpout getKafkaSpout(Properties conf, String topic) {
    SpoutConfig spoutConfig = new SpoutConfig(
            new ZkHosts(conf.getProperty("zk_host"), "/brokers"),
            topic,
            "/brokers",
            conf.getProperty("kafka_consumer_group_id"));
    List<String> zkServers = new ArrayList<String>();
    zkServers.add(conf.getProperty("zk_host"));
    spoutConfig.zkServers = zkServers;
    spoutConfig.zkPort = Integer.valueOf(conf.getProperty("zk_port"));
    spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
    spoutConfig.forceFromStart = true;
    spoutConfig.fetchSizeBytes = 5*1024*1024;
    spoutConfig.bufferSizeBytes = 5*1024*1024;
    storm.kafka.KafkaSpout kafkaSpout = new storm.kafka.KafkaSpout(spoutConfig);
    return kafkaSpout;
}

We use AWS c3.2xlarge machines, Apache Storm 0.9.2-incubating, Apache Kafka 2.9.2-0.8.1.1.

Answer 1

Test Ping and Telnet: make sure each machine that runs storm has connection to all other machines with ping (all workers, nimbus and zookeeper). try pinging via IP, hostname and FQDN and if it doesn't work, edit the hosts (/etc/hosts)files so it does.

also, telnet the machines to check open ports in storm.yaml (6701, 6702). Zookeeper (2181).

in my tested environment, the storm.yaml settings works with the following netty settings:

storm.messaging.netty.buffer_size: 5242880
storm.messaging.netty.client_worker_threads: 1
storm.messaging.netty.max_retries: 100
storm.messaging.netty.max_wait_ms: 1000
storm.messaging.netty.min_wait_ms: 100
storm.messaging.netty.server_worker_threads: 1
storm.messaging.transport: backtype.storm.messaging.netty.Context