### Question:
CommonCrawlIndex { public static FlowDef createCommonCrawlFlowDef(Tap source, Tap sink) { Pipe parsePipe = new Pipe( "exportCommonCrawlWATPipe" ); RegexGenerator splitter=new RegexGenerator(new Fields("json"),"^\\{\"Envelope\".*$"); parsePipe = new Each( parsePipe, new Fields( "line" ), splitter, Fields.RESULTS ); return FlowDef.flowDef() .addSource( parsePipe, source ) .addTailSink( parsePipe, sink ); }  static FlowDef buildFlowDef(Properties properties); static FlowDef createCommonCrawlFlowDef(Tap source, Tap sink);  }### Answer:
@Test public void testCreateCommonCrawlFlowDef() throws Exception { Properties properties = new ConfigReader().renderProperties(CommonCrawlIndexTest.class); String sourcePath = properties.getProperty("inPath"); String sinkPath = properties.getProperty("testCreateCommonCrawlFlowDefOutput"); String sinkValidationPath = properties.getProperty("testCreateCommonCrawlFlowDefOutputValidation"); Tap source = new FileTap(new TextLine(new Fields("line")) ,sourcePath); Tap sink = new FileTap(new TextLine(new Fields("line")) ,sinkPath); FlowDef flowDef = CommonCrawlIndex.createCommonCrawlFlowDef(source, sink); new LocalFlowConnector(properties).connect(flowDef).complete(); Assert.sameContent(sinkPath, sinkValidationPath); }