content: /*
   * Licensed to the Apache Software Foundation (ASF) under one
   * or more contributor license agreements.  See the NOTICE file
   * distributed with this work for additional information
   * regarding copyright ownership.  The ASF licenses this file
   * to you under the Apache License, Version 2.0 (the
   * "License"); you may not use this file except in compliance
   * with the License.  You may obtain a copy of the License at
   *
   *     http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.hadoop.chukwa.analysis.salsa.fsm;
  
  import java.io.BufferedReader;
  import java.io.File;
  import java.io.FileReader;
  import java.io.FilenameFilter;
  import java.io.IOException;
  import java.text.SimpleDateFormat;
  import java.util.ArrayList;
  import java.util.Calendar;
  import java.util.Date;
  import java.util.regex.*;
  
  import org.apache.commons.io.FileUtils;
  import org.apache.commons.logging.Log;
  import org.apache.commons.logging.LogFactory;
  import org.apache.hadoop.chukwa.conf.ChukwaConfiguration;
  import org.apache.hadoop.chukwa.database.TableCreator;
  import org.apache.hadoop.chukwa.datacollection.DataFactory;
  import org.apache.hadoop.chukwa.datacollection.agent.ChukwaAgent;
  import org.apache.hadoop.chukwa.datacollection.agent.ChukwaAgent.AlreadyRunningException;
  import org.apache.hadoop.chukwa.datacollection.connector.http.HttpConnector;
  import org.apache.hadoop.chukwa.datacollection.collector.CaptureWriter;
  import org.apache.hadoop.chukwa.datacollection.collector.servlet.ServletCollector;
  import org.apache.hadoop.chukwa.datacollection.controller.ChukwaAgentController;
  import org.apache.hadoop.chukwa.datacollection.sender.ChukwaHttpSender;
  import org.apache.hadoop.chukwa.datacollection.sender.RetryListOfCollectors;
  import org.apache.hadoop.chukwa.datacollection.test.ConsoleOutConnector;
  import org.apache.hadoop.chukwa.datacollection.writer.PipelineStageWriter;
  import org.apache.hadoop.chukwa.dataloader.MetricDataLoader;
  import org.apache.hadoop.conf.Configuration;
  import org.mortbay.jetty.Connector;
  import org.mortbay.jetty.Server;
  import org.mortbay.jetty.nio.SelectChannelConnector;
  import org.mortbay.jetty.servlet.Context;
  import org.mortbay.jetty.servlet.ServletHolder;
  import org.apache.hadoop.hdfs.MiniDFSCluster;
  import org.apache.hadoop.io.SequenceFile;
  import org.apache.hadoop.mapred.FileInputFormat;
  import org.apache.hadoop.mapred.FileOutputFormat;
  import org.apache.hadoop.mapred.JobConf;
  import org.apache.hadoop.mapred.JobClient;
  import org.apache.hadoop.mapred.JobPriority;
  import org.apache.hadoop.mapred.MiniMRCluster;
  import org.apache.hadoop.mapred.SequenceFileInputFormat;
  import org.apache.hadoop.util.ToolRunner;
  import org.apache.hadoop.chukwa.ChukwaArchiveKey;
  import org.apache.hadoop.chukwa.ChunkImpl;
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.fs.*;
  import org.apache.hadoop.chukwa.extraction.demux.ChukwaRecordOutputFormat;
  import org.apache.hadoop.chukwa.extraction.demux.ChukwaRecordPartitioner;
  import org.apache.hadoop.chukwa.extraction.demux.Demux;
  import org.apache.hadoop.chukwa.extraction.engine.ChukwaRecord;
  import org.apache.hadoop.chukwa.extraction.engine.ChukwaRecordKey;
  import org.apache.hadoop.chukwa.util.DatabaseWriter;
  import org.apache.hadoop.chukwa.util.ExceptionUtil;
  import org.apache.hadoop.chukwa.analysis.salsa.fsm.*;
  
  import junit.framework.TestCase;
  
  public class TestFSMBuilder extends TestCase {
    private static Log log = LogFactory.getLog(TestFSMBuilder.class);
  
    int LINES = 10000;
    int THREADS = 2;
    private MiniDFSCluster dfs = null;
    int NUM_HADOOP_SLAVES = 4;
    private FileSystem fileSys = null;
    private MiniMRCluster mr = null;
    private Server jettyCollector = null;
    private ChukwaAgent agent = null;
    private HttpConnector conn = null;
    private ChukwaHttpSender sender = null;
    private int agentPort = 9093;
    private int collectorPort = 9990;
    private static final String dataSink = "/demux/input";
    private static final String fsmSink = "/analysis/salsafsm";
    private static Path DEMUX_INPUT_PATH = null;
    private static Path DEMUX_OUTPUT_PATH = null;
    private static Path FSM_OUTPUT_PATH = null;
    private ChukwaConfiguration conf = new ChukwaConfiguration();
    private static SimpleDateFormat day = new java.text.SimpleDateFormat("yyyyMMdd_HH_mm");
    private static String cluster = "demo";
    long[] timeWindow = {7, 30, 91, 365, 3650};
    long current = 1244617200000L;  // 2009-06-10
    private String testBuildDir = System.getProperty("test.build.data", "/tmp");
    private File dfsDataDir = new File(testBuildDir+"/dfs");
  
    public void setUp() {
      // Startup HDFS cluster - stored collector-ed JobHistory chunks
      // Startup MR cluster - run Demux, FSMBuilder
      // Startup collector
      // Startup agent
      
      System.out.println("In setUp()");
      try {
        System.setProperty("hadoop.log.dir", System.getProperty(
            "test.build.data", "/tmp"));
      } catch (Exception e) {
        e.printStackTrace();
        fail("Could not set up: " + e.toString());
      }  
  
      // Startup HDFS cluster - stored collector-ed JobHistory chunks
      try {
        if(dfsDataDir.exists()) {
          FileUtils.deleteDirectory(dfsDataDir);        
        }
        dfs = new MiniDFSCluster(conf, NUM_HADOOP_SLAVES, true, null);
        fileSys = dfs.getFileSystem();
        DEMUX_INPUT_PATH = new Path(fileSys.getUri().toString()+File.separator+dataSink);          
        DEMUX_OUTPUT_PATH = new Path(fileSys.getUri().toString()+File.separator+"/demux/output");
      } catch(Exception e) {
        e.printStackTrace();
        fail("Fail to startup HDFS cluster.");      
      }
      // Startup MR Cluster
      try {
        mr = new MiniMRCluster(NUM_HADOOP_SLAVES, fileSys.getUri()
            .toString(), 1);
      } catch(Exception e) {
        fail("Fail to startup Map/reduce cluster.");
      }
      // Startup collector
      try {
        // Configure Collector
        conf.set("chukwaCollector.chunkSuppressBufferSize", "10");
        conf.set("writer.hdfs.filesystem",fileSys.getUri().toString());
        conf.set("chukwaCollector.outputDir",dataSink);
        conf.set("chukwaCollector.rotateInterval", "10000");
        
        // Set up jetty connector
        SelectChannelConnector jettyConnector = new SelectChannelConnector();
        jettyConnector.setLowResourcesConnections(THREADS-1);
        jettyConnector.setLowResourceMaxIdleTime(1500);
        jettyConnector.setPort(collectorPort);
        
        // Set up jetty server proper, using connector
        jettyCollector = new Server(collectorPort);
        Context root = new Context(jettyCollector, "/", Context.SESSIONS);
        root.addServlet(new ServletHolder(new ServletCollector(conf)), "/*");
        jettyCollector.start();
        jettyCollector.setStopAtShutdown(true);
        Thread.sleep(10000);
      } catch(Exception e) {
        fail("Fail to startup collector.");
      }
  
      // Startup agent
      try {
        // Configure Agent
        conf.set("chukwaAgent.tags", "cluster=\"demo\"");
        DataFactory.getInstance().addDefaultTag(conf.get("chukwaAgent.tags", "cluster=\"unknown\""));
        conf.set("chukwaAgent.checkpoint.dir", System.getenv("CHUKWA_DATA_DIR")+File.separator+"tmp");
        conf.set("chukwaAgent.checkpoint.interval", "10000");
        int portno = conf.getInt("chukwaAgent.control.port", agentPort);
        agent = ChukwaAgent.getAgent();
        agent.start();
        conn = new HttpConnector(agent, "http://localhost:"+collectorPort+"/chukwa");
        conn.start();      
        sender = new ChukwaHttpSender(conf);
        ArrayList<String> collectorList = new ArrayList<String>();
        collectorList.add("http://localhost:"+collectorPort+"/chukwa");
        sender.setCollectors(new RetryListOfCollectors(collectorList, conf));
      } catch (AlreadyRunningException e) {
        fail("Chukwa Agent is already running");
      }
      System.out.println("Done setUp().");
    }
  
    public String readFile(File aFile) {
      StringBuffer contents = new StringBuffer();
      try {
        BufferedReader input = new BufferedReader(new FileReader(aFile));
        try {
          String line = null; // not declared within while loop
          while ((line = input.readLine()) != null) {
            contents.append(line);
            contents.append(System.getProperty("line.separator"));
          }
        } finally {
          input.close();
        }
      } catch (IOException ex) {
        ex.printStackTrace();
      }
      return contents.toString();
    }
  
    public void tearDown() {
      FileSystem fs;
      System.out.println("In tearDown()");
      try {      
  
        fs = dfs.getFileSystem();
        fs.delete(DEMUX_OUTPUT_PATH, true);
  
        agent.shutdown();
        conn.shutdown();
        jettyCollector.stop();
        mr.shutdown();
        dfs.shutdown();
        if(dfsDataDir.exists()) {
          FileUtils.deleteDirectory(dfsDataDir);        
        }
        Thread.sleep(2000);
      } catch(Exception e) {
        e.printStackTrace();
        fail(e.toString());
      }
      System.out.println("Done tearDown()");
    }
  
    /**
     * Performs tasks common to all tests
     * Sets up agent to collect samples of the 2 log types in use
     * (job history logs via JobLog and clienttrace via ClientTrace log types)
     * Calls Demux to process the logs
     */
    protected void initialTasks () {
      System.out.println("In initialTasks()");
      try {
        // Test Chukwa Agent Controller and Agent Communication
        ChukwaAgentController cli = new ChukwaAgentController("localhost", agentPort);
        String[] source = new File(System.getenv("CHUKWA_DATA_DIR") + File.separator + "log").list(new FilenameFilter() {
          public boolean accept(File dir, String name) {
            return name.endsWith(".log");
          }
        });
        System.out.println(System.getenv("CHUKWA_DATA_DIR") + File.separator + "log");
        for(String fname : source) {
          if (!(fname.equals("JobHistory.log") || fname.equals("ClientTrace.log"))) {
            continue;
          }
          StringBuilder fullPath = new StringBuilder();
          fullPath.append(System.getProperty("CHUKWA_DATA_DIR"));
          fullPath.append(File.separator);
          fullPath.append("log");
          fullPath.append(File.separator);        
          fullPath.append(fname);
          String recordType = fname.substring(0,fname.indexOf("."));
          String adaptorId = cli.add(
            "org.apache.hadoop.chukwa.datacollection.adaptor.filetailer.CharFileTailingAdaptorUTF8NewLineEscaped", 
            recordType, "0 " + fullPath.toString(), 0);
          assertNotNull(adaptorId);
          Thread.sleep(2000);
        }
        cli.removeAll();
        Thread.sleep(30000);
      } catch (Exception e) {
        e.printStackTrace();
        fail(e.toString());
      }
      
      // Test Data Sink files written by Collector    
      Path demuxDir = new Path(dataSink+"/*");
      FileSystem fs;
      try {
        fs = dfs.getFileSystem();
        FileStatus[] events = fs.globStatus(demuxDir);
        log.info("Number of data sink files written:"+events.length);
        assertTrue(events.length!=0);
      } catch (IOException e) {
        e.printStackTrace();
        fail("File System Error.");
      }
      
      // Test Demux    
      log.info("Testing demux");
      try {
        //ChukwaConfiguration conf = new ChukwaConfiguration();
        System.setProperty("hadoop.log.dir", System.getProperty(
            "test.build.data", "/tmp"));
      
        String[] sortArgs = { DEMUX_INPUT_PATH.toString(), DEMUX_OUTPUT_PATH.toString() };
        //      JobConf job = mr.createJobConf();
        JobConf job = new JobConf(new ChukwaConfiguration(), Demux.class);
        job.addResource(System.getenv("CHUKWA_CONF_DIR")+File.separator+"chukwa-demux-conf.xml");
        job.setJobName("Chukwa-Demux_" + day.format(new Date()));
        job.setInputFormat(SequenceFileInputFormat.class);
        job.setMapperClass(Demux.MapClass.class);
        job.setPartitionerClass(ChukwaRecordPartitioner.class);
        job.setReducerClass(Demux.ReduceClass.class);
  
        job.setOutputKeyClass(ChukwaRecordKey.class);
        job.setOutputValueClass(ChukwaRecord.class);
        job.setOutputFormat(ChukwaRecordOutputFormat.class);
        job.setJobPriority(JobPriority.VERY_HIGH);
        job.setNumMapTasks(2);
        job.setNumReduceTasks(1);
        Path input = new Path(fileSys.getUri().toString()+File.separator+dataSink+File.separator+"*.done");
        FileInputFormat.setInputPaths(job, input);
        FileOutputFormat.setOutputPath(job, DEMUX_OUTPUT_PATH);
        String[] jars = new File(System.getenv("CHUKWA_HOME")).list(new FilenameFilter() {
          public boolean accept(File dir, String name) {
            return name.endsWith(".jar");
          }
        });
        job.setJar(System.getenv("CHUKWA_HOME")+File.separator+jars[0]);
        //assertEquals(ToolRunner.run(job, new Demux(), sortArgs), 0);
        JobClient.runJob(job);
      } catch (Exception e) {
        fail(ExceptionUtil.getStackTrace(e));
      }
  
      System.out.println("Done initialTasks()");
    }
  
    public void testFSMBuilder_JobHistory020 () {
      initialTasks();
      // Test FSMBuilder (job history only)
      log.info("Testing FSMBuilder (Job History only)");
      System.out.println("In JobHistory020");
      // Run FSMBuilder on Demux output
      try {
        JobConf job = new JobConf(new ChukwaConfiguration(), FSMBuilder.class);
        job.addResource(System.getenv("CHUKWA_CONF_DIR")+File.separator+"chukwa-demux-conf.xml");
        job.setJobName("Chukwa-FSMBuilder_" + day.format(new Date()));
        job.setMapperClass(JobHistoryTaskDataMapper.class);
        job.setPartitionerClass(FSMIntermedEntryPartitioner.class);
        job.setReducerClass(FSMBuilder.FSMReducer.class);
        job.setMapOutputValueClass(FSMIntermedEntry.class);
        job.setMapOutputKeyClass(ChukwaRecordKey.class);
  
        job.setInputFormat(SequenceFileInputFormat.class);
        job.setOutputKeyClass(ChukwaRecordKey.class);
        job.setOutputValueClass(ChukwaRecord.class);
        job.setOutputFormat(ChukwaRecordOutputFormat.class);
        job.setNumReduceTasks(1);
  
        Path inputPath = new Path(DEMUX_OUTPUT_PATH.toString()+File.separator+"/*/*/TaskData*.evt");
        this.FSM_OUTPUT_PATH = new Path(fileSys.getUri().toString()+File.separator+fsmSink);
  
        FileInputFormat.setInputPaths(job, inputPath);
        FileOutputFormat.setOutputPath(job, FSM_OUTPUT_PATH);
  
        String[] jars = new File(System.getenv("CHUKWA_HOME")).list(new FilenameFilter() {
          public boolean accept(File dir, String name) {
            return name.endsWith(".jar");
          }
        });
        job.setJar(System.getenv("CHUKWA_HOME")+File.separator+jars[0]);
        JobClient.runJob(job);
      } catch (Exception e) {
        fail("Error running FSMBuilder: "+e.toString());
      }
      System.out.println("Done running FSMBuilder; Checking results");
    
      // Check FSMBuilder output by reading the sequence file(s) generated
      // Hard-coded to check the contents of test/samples/JobLog.log
      try {
  
        Pattern task_id_pat = Pattern.compile("attempt_[0-9]*_[0-9]*_[mr]_([0-9]*)_[0-9]*");
  
        ChukwaRecordKey key = new ChukwaRecordKey();
        ChukwaRecord record = new ChukwaRecord();
  
        // initialize data structures for checking FSM
        // should see 10 maps, 8 reduces
        boolean mapSeen[] = new boolean[10];
        boolean reduceSeen[] = new boolean[8];
        boolean reduceShuffleSeen[] = new boolean[8];
        boolean reduceSortSeen[] = new boolean[8];
        boolean reduceReducerSeen[] = new boolean[8];
        for (int i = 0; i < 10; i++) mapSeen[i] = false;
        for (int i = 0; i < 8; i++) { 
          reduceSeen[i] = false;
          reduceShuffleSeen[i] = false;
          reduceSortSeen[i] = false;
          reduceReducerSeen[i] = false;
        }
  
        Path fsm_outputs = new Path(FSM_OUTPUT_PATH.toString()+File.separator+
          "/*/MAPREDUCE_FSM/MAPREDUCE_FSM*.evt");
        FileStatus [] files;
        files = fileSys.globStatus(fsm_outputs);
        int count = 0;
  
        for (int i = 0; i < files.length; i++) {
          SequenceFile.Reader r = new SequenceFile.Reader(fileSys, files[i].getPath(), conf);
          System.out.println("Processing files " + files[i].getPath().toString());
          while (r.next(key, record)) {
            String state_name = record.getValue("STATE_NAME");
            String task_id = record.getValue("TASK_ID");
            
            Matcher m = task_id_pat.matcher(task_id);
            if (!m.matches()) {
              continue;
            }
            String tasknum_string = m.group(1);
            if (tasknum_string == null) {
              continue;
            }
            int tasknum = Integer.parseInt(tasknum_string);
    
            if (state_name.equals("MAP")) {
              assertTrue("Map sequence number should be < 10",tasknum < 10);
              mapSeen[tasknum] = true;
            } else if (state_name.equals("REDUCE")) {
              assertTrue("Reduce sequence number should be < 8",tasknum < 8);
              reduceSeen[tasknum] = true;
            } else if (state_name.equals("REDUCE_SHUFFLEWAIT")) {
              assertTrue("Reduce sequence number should be < 8",tasknum < 8);
              reduceShuffleSeen[tasknum] = true;
            } else if (state_name.equals("REDUCE_SORT")) {
              assertTrue("Reduce sequence number should be < 8",tasknum < 8);
              reduceSortSeen[tasknum] = true;
            } else if (state_name.equals("REDUCE_REDUCER")) {
              assertTrue("Reduce sequence number should be < 8",tasknum < 8);
              reduceReducerSeen[tasknum] = true;
            }
            count++;
          }
        }
        System.out.println("Processed " + count + " records.");
        assertTrue("Total number of states is 42 - 10 maps + (8 reduces * 4)",count == 42);  
  
        // We must have seen all 10 maps and all 8 reduces; 
        // check for that here
        boolean passed = true;
        for (int i = 0; i < 10; i++) passed &= mapSeen[i];
        for (int i = 0; i < 8; i++) {
          passed &= reduceSeen[i];
          passed &= reduceShuffleSeen[i];
          passed &= reduceSortSeen[i];
          passed &= reduceReducerSeen[i];
        }
  
        assertTrue("Seen all Maps and Reduces in generated states.",passed);
  
      } catch (Exception e) {
        fail("Error checking FSMBuilder output: "+e.toString());
      } 
      
    }
  
    public void testFSMBuilder_ClientTrace020 () {
      initialTasks();
      // Test FSMBuilder (job history only)
      log.info("Testing FSMBuilder (ClientTrace only)");
      System.out.println("In ClientTrace020");
      // Run FSMBuilder on Demux output
      try {
        // Process TaskTracker shuffle clienttrace entries first
        JobConf job = new JobConf(new ChukwaConfiguration(), FSMBuilder.class);
        job.addResource(System.getenv("CHUKWA_CONF_DIR")+File.separator+"chukwa-demux-conf.xml");
        job.setJobName("Chukwa-FSMBuilder_" + day.format(new Date()));
        job.setMapperClass(TaskTrackerClientTraceMapper.class);
        job.setPartitionerClass(FSMIntermedEntryPartitioner.class);
        job.setReducerClass(FSMBuilder.FSMReducer.class);
        job.setMapOutputValueClass(FSMIntermedEntry.class);
        job.setMapOutputKeyClass(ChukwaRecordKey.class);
  
        job.setInputFormat(SequenceFileInputFormat.class);
        job.setOutputKeyClass(ChukwaRecordKey.class);
        job.setOutputValueClass(ChukwaRecord.class);
        job.setOutputFormat(ChukwaRecordOutputFormat.class);
        job.setNumReduceTasks(1);
  
        Path inputPath = new Path(DEMUX_OUTPUT_PATH.toString()+File.separator+"/*/*/ClientTraceDetailed*.evt");
        Path fsmOutputPath1 = new Path(fileSys.getUri().toString()+File.separator+fsmSink+"1");
  
        FileInputFormat.setInputPaths(job, inputPath);
        FileOutputFormat.setOutputPath(job, fsmOutputPath1);
  
        String[] jars = new File(System.getenv("CHUKWA_HOME")).list(new FilenameFilter() {
          public boolean accept(File dir, String name) {
            return name.endsWith(".jar");
          }
        });
        job.setJar(System.getenv("CHUKWA_HOME")+File.separator+jars[0]);
        JobClient.runJob(job);
        System.out.println("Processed TaskTracker ClientTrace.");
  
        // Process DataNode clienttrace entries
        job = new JobConf(new ChukwaConfiguration(), FSMBuilder.class);
        job.addResource(System.getenv("CHUKWA_CONF_DIR")+File.separator+"chukwa-demux-conf.xml");
        job.setJobName("Chukwa-FSMBuilder_" + day.format(new Date()));
        job.setMapperClass(DataNodeClientTraceMapper.class);
        job.setPartitionerClass(FSMIntermedEntryPartitioner.class);
        job.setReducerClass(FSMBuilder.FSMReducer.class);
        job.setMapOutputValueClass(FSMIntermedEntry.class);
        job.setMapOutputKeyClass(ChukwaRecordKey.class);
  
        job.setInputFormat(SequenceFileInputFormat.class);
        job.setOutputKeyClass(ChukwaRecordKey.class);
        job.setOutputValueClass(ChukwaRecord.class);
        job.setOutputFormat(ChukwaRecordOutputFormat.class);
        job.setNumReduceTasks(1);
  
        inputPath = new Path(DEMUX_OUTPUT_PATH.toString()+File.separator+"/*/*/ClientTraceDetailed*.evt");
        Path fsmOutputPath2 = new Path(fileSys.getUri().toString()+File.separator+fsmSink+"2");
  
        FileInputFormat.setInputPaths(job, inputPath);
        FileOutputFormat.setOutputPath(job, fsmOutputPath2);
  
        jars = new File(System.getenv("CHUKWA_HOME")).list(new FilenameFilter() {
          public boolean accept(File dir, String name) {
            return name.endsWith(".jar");
          }
        });
        job.setJar(System.getenv("CHUKWA_HOME")+File.separator+jars[0]);
        JobClient.runJob(job);
        System.out.println("Processed DataNode ClientTrace.");
  
      } catch (Exception e) {
        fail("Error running FSMBuilder: "+e.toString());
      }
      System.out.println("Done running FSMBuilder; Checking results");
  
      try {
        Path fsm_outputs = new Path(fileSys.getUri().toString()+File.separator+
          fsmSink + "*/*/*/*.evt");
        FileStatus [] files;
        files = fileSys.globStatus(fsm_outputs);
        int count = 0;
        int numHDFSRead = 0, numHDFSWrite = 0, numShuffles = 0;
        ChukwaRecordKey key = new ChukwaRecordKey();
        ChukwaRecord record = new ChukwaRecord();
  
        for (int i = 0; i < files.length; i++) {
          SequenceFile.Reader r = new SequenceFile.Reader(fileSys, files[i].getPath(), conf);
          System.out.println("Processing files " + files[i].getPath().toString());
          while (r.next(key, record)) {
            String state_name = record.getValue("STATE_NAME");
    
            if (state_name.equals("READ_LOCAL") || state_name.equals("READ_REMOTE")) 
            {
              numHDFSRead++;
            } else if (state_name.equals("WRITE_LOCAL") || state_name.equals("WRITE_REMOTE") 
                || state_name.equals("WRITE_REPLICATED")) 
            {
              numHDFSWrite++;
            } else if (state_name.equals("SHUFFLE_LOCAL") || state_name.equals("SHUFFLE_REMOTE")) 
            {
              numShuffles++;
            }
            count++;
          }
        }
        System.out.println("Processed " + count + " records."); 
        System.out.println("HDFSRD: " + numHDFSRead + " HDFSWR: " + numHDFSWrite + " SHUF: " + numShuffles);
        assertTrue("Number of HDFS reads", numHDFSRead == 10);
        assertTrue("Number of HDFS writes", numHDFSWrite == 8);      
        assertTrue("Number of shuffles", numShuffles == 80);
  
      } catch (Exception e) {
        fail("Error checking FSMBuilder results: " + e.toString());
      }      
    }
  
  }
  
  