package storm.benchmark.trident.operation;

import backtype.storm.tuple.Values;
import storm.trident.operation.BaseFunction;
import storm.trident.operation.TridentCollector;
import storm.trident.tuple.TridentTuple;

public class WordSplit extends BaseFunction {

  private static final long serialVersionUID = -8605358179216330897L;

  public static String[] splitSentence(String sentence) {
    if (sentence != null) {
      return sentence.split("\\s+");
    }
    return null;
  }

  @Override
  public void execute(TridentTuple tuple, TridentCollector collector) {
    String sentence = tuple.getString(0);
    if (sentence != null) {
      for (String word : splitSentence(sentence)) {
        collector.emit(new Values(word));
      }
    }
  }
}

@@ -73,16 +73,48 @@
 Values;%0A
+import org.apache.log4j.Logger;%0A
 import s
@@ -288,16 +288,87 @@
 tion %7B%0A%0A
+  private static final Logger LOG = Logger.getLogger(WordSplit.class);%0A
   privat
@@ -707,16 +707,71 @@
 ing(0);%0A
+    LOG.debug(%22receive sentence: '%22 + sentence + %22'%22);%0A
     if (

add debug log