TrapPlatformTest.java

  @Test
  public void testTrapTapSourceSinkCopy() throws Exception
    {
    getPlatform().copyFromLocal( inputFileApache );

    Scheme scheme = getPlatform().getTestFailScheme();

    Tap source = getPlatform().getTap( scheme, inputFileApache, SinkMode.KEEP );

    Pipe pipe = new Pipe( "map" );

    Tap sink = getPlatform().getTap( scheme, getOutputPath( "trapsourcesinkcopy/sink" ), SinkMode.REPLACE );
    Tap trap = getPlatform().getTextFile( new Fields( "line" ), getOutputPath( "trapsourcesinkcopy/trap" ), SinkMode.REPLACE );

    Map<Object, Object> properties = getProperties();

    // compensate for running in cluster mode
    properties.put( "mapred.map.tasks", 1 );
    properties.put( "mapred.reduce.tasks", 1 );

    FlowDef flowDef = FlowDef.flowDef()
      .addSource( pipe, source )
      .addTailSink( pipe, sink )
      .addTrap( pipe, trap );

    Flow flow = getPlatform().getFlowConnector( properties ).connect( flowDef );

    flow.complete();

    validateLength( flow.openTapForRead( getPlatform().getTextFile( sink.getIdentifier() ) ), 9 );
    validateLength( flow.openTrap(), 2, Pattern.compile( "bad data" ) ); // confirm the payload is written
    }