Setting up a new stream for warc bolt fails

I'm trying to setup a new stream to connect a Tika bolt to a warc bolt.

import com.digitalpebble.stormcrawler.tika.ParserBolt;
import com.digitalpebble.stormcrawler.warc.WARCHdfsBolt;

builder.setBolt("tika", new ParserBolt(), numWorkers)
  .localOrShuffleGrouping("shunt","tika");

WARCHdfsBolt warcbolt = getWarcBolt("XX");

builder.setBolt("warc", warcbolt, numWorkers)
  .localOrShuffleGrouping("tika",  "warc");

In the Tika definition I have modified the outputDeclarerFields function as follows to define my new "warc" stream:

@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
  declarer.declare(new Fields("url", "content", "metadata", "text"));
  declarer.declareStream(StatusStreamName, new Fields("url", "metadata", "status"));
  declarer.declareStream("warc",   new Fields("url", "content", "metadata", "text"));
}

However, when I launch the topology in local mode I get:

14308 [main] WARN o.a.s.d.s.Slot - SLOT debian8:1027 Starting in state EMPTY - assignment null 14308 [main] WARN o.a.s.d.s.Slot - SLOT debian8:1028 Starting in state EMPTY - assignment null 14308 [main] WARN o.a.s.d.s.Slot - SLOT debian8:1029 Starting in state EMPTY - assignment null 14309 [main] INFO o.a.s.l.AsyncLocalizer - Cleaning up unused topologies in /tmp/a1e3b7f5-e251-40ae-a032-b0839ca103c8/supervisor/stormdist 14318 [main] INFO o.a.s.d.s.Supervisor - Starting supervisor with id f42c64cd-7c36-40ab-9f85-4b7751ed2d6a at host debian8. 15030 [main] WARN o.a.s.d.nimbus - Topology submission exception. (topology name='xxCrawler') #error { :cause nil :via [{:type org.apache.storm.generated.InvalidTopologyException :message nil
:at [org.apache.storm.daemon.common$validate_structure_BANG_ invoke common.clj 185]}] :trace [[org.apache.storm.daemon.common$validate_structure_BANG_ invoke common.clj 185]
[org.apache.storm.daemon.common$system_topology_BANG_ invoke common.clj 378]
[org.apache.storm.daemon.nimbus$mk_reified_nimbus$reify__10782 submitTopologyWithOpts nimbus.clj 1694]
[org.apache.storm.daemon.nimbus$mk_reified_nimbus$reify__10782 submitTopology nimbus.clj 1726]
[sun.reflect.NativeMethodAccessorImpl invoke0 NativeMethodAccessorImpl.java -2]
[sun.reflect.NativeMethodAccessorImpl invoke NativeMethodAccessorImpl.java 62]
[sun.reflect.DelegatingMethodAccessorImpl invoke DelegatingMethodAccessorImpl.java 43] [java.lang.reflect.Method invoke Method.java 498] [clojure.lang.Reflector invokeMatchingMethod Reflector.java 93] [clojure.lang.Reflector invokeInstanceMethod Reflector.java 28] [org.apache.storm.testing$submit_local_topology invoke testing.clj 310]
[org.apache.storm.LocalCluster$_submitTopology invoke LocalCluster.clj 49] [org.apache.storm.LocalCluster submitTopology nil -1]
[com.digitalpebble.stormcrawler.ConfigurableTopology submit ConfigurableTopology.java 76]
[com.digitalpebble.stormcrawler.ConfigurableTopology submit ConfigurableTopology.java 65] [xx.xx.xx.xx.xxTopology run xxTopology.java 111]
[com.digitalpebble.stormcrawler.ConfigurableTopology start ConfigurableTopology.java 50] [xx.xx.xx.xx.xxTopology main xxTopology.java 53]]} 15035 [main] ERROR o.a.s.s.o.a.z.s.NIOServerCnxnFactory - Thread Thread[main,5,main] died org.apache.storm.generated.InvalidTopologyException: null at org.apache.storm.daemon.common$validate_structure_BANG_.invoke(common.clj:185) ~[storm-core-1.1.0.jar:1.1.0] at org.apache.storm.daemon.common$system_topology_BANG_.invoke(common.clj:378) ~[storm-core-1.1.0.jar:1.1.0] at org.apache.storm.daemon.nimbus$mk_reified_nimbus$reify__10782.submitTopologyWithOpts(nimbus.clj:1694) ~[storm-core-1.1.0.jar:1.1.0] at org.apache.storm.daemon.nimbus$mk_reified_nimbus$reify__10782.submitTopology(nimbus.clj:1726) ~[storm-core-1.1.0.jar:1.1.0] at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:1.8.0_131] at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[?:1.8.0_131] at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[?:1.8.0_131] at java.lang.reflect.Method.invoke(Method.java:498) ~[?:1.8.0_131] at clojure.lang.Reflector.invokeMatchingMethod(Reflector.java:93) ~[clojure-1.7.0.jar:?] at clojure.lang.Reflector.invokeInstanceMethod(Reflector.java:28) ~[clojure-1.7.0.jar:?] at org.apache.storm.testing$submit_local_topology.invoke(testing.clj:310) ~[storm-core-1.1.0.jar:1.1.0] at org.apache.storm.LocalCluster$_submitTopology.invoke(LocalCluster.clj:49) ~[storm-core-1.1.0.jar:1.1.0] at org.apache.storm.LocalCluster.submitTopology(Unknown Source) ~[storm-core-1.1.0.jar:1.1.0] at com.digitalpebble.stormcrawler.ConfigurableTopology.submit(ConfigurableTopology.java:76) ~[xx-crawler-1.1.jar:?] at com.digitalpebble.stormcrawler.ConfigurableTopology.submit(ConfigurableTopology.java:65) ~[xx-1.1.jar:?] at xx.xx.xx.xx.xxTopology.run(xxTopology.java:111) ~[xx-crawler-1.1.jar:?] at com.digitalpebble.stormcrawler.ConfigurableTopology.start(ConfigurableTopology.java:50) ~[xx-crawler-1.1.jar:?] at xx.xx.xx.xx.xxTopology.main(xxTopology.java:53) ~[xx-crawler-1.1.jar:?]

Any help would be highly appreciated!!

Note that if I use the StatusStreamName ("status") stream to connect the tika and warc bolts it works fine.

Thank you,

Etienne

Solution

WARCs are generated from the raw, non-parsed content. You should connect the WARC to the output of the Fetcher instead of the Parser bolt.

You don't need to declare a new stream just for the warc, you could simply connect the warc bolt to the default stream coming out of the Tika bolt.

I see in your code

import com.digitalpebble.stormcrawler.tika.ParserBolt;

which would indicate that you are relying on the default implementation (which does not generate a 'warc' stream). Could you have forgotten to replace that with your modified implementation?