Spring integration file read issues

Sounds like spring integeration is amazing , I am new in Spring Integeration.I have one requirement Using spring integeration

1. read a txt file (from Source folder)
2. do some validation
3. if validation is success -write into sucess file (in sucess folder)
4. If the validation is fail -write into failure file (in error folder)

My config file is like this

     <?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans"
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:si="http://www.springframework.org/schema/integration"
        xmlns:file="http://www.springframework.org/schema/integration/file"
        xsi:schemaLocation="http://www.springframework.org/schema/beans
                http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
                http://www.springframework.org/schema/integration
                http://www.springframework.org/schema/integration/spring-integration-1.0.xsd
                http://www.springframework.org/schema/integration/file
                http://www.springframework.org/schema/integration/file/spring-integration-file-1.0.xsd">

        <bean id="checkCSVReader"
            class="com.check.wrapper">
            <property name="pzMapXML" value="classpath:sampleFileFormat.xml" />
        </bean>

        <bean id="checkTrasnFomer"
            class="com.check.checkTransfomer">
            <property name="wrapper" ref="checkCSVReader" />
        </bean>

        <bean id="fileErrorProcessor"
            class="com.check.ErrorChannelWriter">
        </bean>
        <bean id="listToStringTrans"
            class="com.check.ListToStringTransfomer"></bean>


    <bean id="validation"
            class="com.check.Validation"/>

        <file:inbound-channel-adapter directory="file://D:\check\soruce"   prevent-duplicates="false" 
            auto-create-directory="true" channel="readChannel" >
            <si:poller id="Poller">
                <si:interval-trigger interval="10000" />
            </si:poller>
        </file:inbound-channel-adapter>

        <si:channel id="readChannel" />

        <si:chain input-channel="readChannel" output-channel="processChannel">
            <si:header-enricher error-channel="errorFile" />
            <file:file-to-string-transformer />
            <si:transformer ref="checkTrasnFomer" method="transform" />
            <si:service-activator ref="validation"
                method="validate" />
        </si:chain>

        <si:channel id="processChannel" />

        <si:transformer ref="listToStringTrans" method="transformList"
            input-channel="processChannel" output-channel="finalOut" />

        <si:channel id="finalOut" />

        <file:outbound-channel-adapter id="checkSuccFileOutBound"
            auto-create-directory="true" delete-source-files="true"
            directory="file://D:\check\success" channel="finalOut">
        </file:outbound-channel-adapter>

        <si:channel id="errorFile" />

        <si:transformer ref="fileErrorProcessor"
            input-channel="errorFile" output-channel="errorChannel" method="transformError" />

        <file:outbound-channel-adapter id="errorChannel"
            directory="file://D:\check\error" delete-source-files="true"
             />

        <si:channel id="checkFileErr" />
    </beans>

my checkFlatPackCVSParserWrapper class is

    public class checkFlatPackCVSParserWrapper {
        private static final Log LOG = LogFactory.getLog("checkFlatPackCVSParserWrapper");
        private Resource pzMapXML;
        private char delimiter = ',';
        private char qualifier = '"';
        private boolean ignoreFirstRecord = false;

        public Resource getPzMapXML() {
            return pzMapXML;
        }
        public void setPzMapXML(Resource pzMapXML) {
            this.pzMapXML = pzMapXML;
        }
        public char getDelimiter() {
            return delimiter;
        }
        public void setDelimiter(char delimiter) {
            this.delimiter = delimiter;
        }
        public char getQualifier() {
            return qualifier;
        }
        public void setQualifier(char qualifier) {
            this.qualifier = qualifier;
        }
        public boolean isIgnoreFirstRecord() {
            return ignoreFirstRecord;
        }
        public void setIgnoreFirstRecord(boolean ignoreFirstRecord) {
            this.ignoreFirstRecord = ignoreFirstRecord;
        }

        public Parser getParser(String csv) {
            if(LOG.isDebugEnabled())
                LOG.debug("getParser: " + csv);

            Parser result = null;
            try {
                result = DefaultParserFactory.getInstance().newDelimitedParser(
                        pzMapXML.getInputStream(), //xml column mapping
                        new ByteArrayInputStream(csv.getBytes()),  //txt file to parse
                        delimiter, //delimiter
                        qualifier, //text qualfier
                        ignoreFirstRecord);

            }catch (Exception e) {
                if(LOG.isDebugEnabled())
                   LOG.debug("Unable to read file:  " + e );
                throw new RuntimeException("File Parse exceptio开发者_JAVA百科n");
            }   
            return result;
        }
    }

sampleFileFormat.xml is

    <?xml version="1.0" encoding="UTF-8"?>
    <!DOCTYPE PZMAP SYSTEM  "flatpack.dtd" >
    <PZMAP>
        <COLUMN name="FIRSTNAME" />
        <COLUMN name="LASTNAME" />
    </PZMAP> 


 and checkTransfomer is 


    public class checkTransfomer {
        private static final Log LOG = LogFactory.getLog(checkTransfomer.class);
        private CheckFlatPackCVSParserWrapper wrapper;

        public String transform(String csv) {
            Parser parser = wrapper.getParser(csv);
            if(LOG.isDebugEnabled()) {
                LOG.debug("Parser is: " + parser);
            }        
            DataSet ds = parser.parse();
            ArrayList<Check> list = new ArrayList<Check>();
            while(ds.next()) {
                Check check= new Check();
                check.setFirstName(ds.getString("FIRSTNAME"));
                check.setLastName(ds.getString("LASTNAME"));
                if(LOG.isDebugEnabled()) {
                    LOG.debug("Bean value is: " + bean);
                }        
                list.add(bean);            
            }
            if(LOG.isDebugEnabled()) {
                LOG.debug("Records fetched is: " + list.size());
            }        
            return list.toString();
        }

        public CheckFlatPackCVSParserWrapper getWrapper() {
            return wrapper;
        }

        public void setWrapper(CheckFlatPackCVSParserWrapper wrapper) {
            this.wrapper = wrapper;
        }

And my ErrorChannelWriter is

    public class ErrorChannelWriter {

        public static final Log LOG = LogFactory.getLog(ErrorChannelWriter.class);

        public Message<?> transformError(ErrorMessage errorMessage) {
            if (LOG.isDebugEnabled()) {
                LOG.debug("Transforming errorMessage is: " + errorMessage);
            }
            return ((MessagingException) errorMessage.getPayload())
                    .getFailedMessage();
        }
        }

and my validagtion class is

  com.check.Validation

  public class Validation 
  {

   void validation(CheckCheck)
   {
  if(Check.getFirstName().equals("maya"))
  {
  throw new RuntimeException("Name Already exist");
  }



  }

  }

and my ListToStringTransfomer is

      public class ListToStringTransfomer {
        private static final Log LOG=LogFactory.getLog(ListToStringTransfomer.class);


        public String transformList(List<IssueAppBean> list) {
            return list.toString();
        }

    }

and my file containing two fields

> maya,kanana 

here if my file format is wrong ,then that records is moving into error file , but if my validation class get fail (here ex: i am checking my first name is equal to maya then i am throwing runtime exception ), that record not moving into error file. how i can acheieve my task suggestions will be appericated