dir-spooler-protolib/src/main/java/com/streamsets/pipeline/lib/dirspooler/SpoolDirRunnable.java
public String generateBatch(WrappedFile file, String offset, int maxBatchSize, BatchMaker batchMaker) throws
      StageException,
      BadSpoolFileException {
    if (offset == null) {
      offset = "0";
    }
    String sourceFile = file.getFileName();
    try {
      if (parser == null) {
        parser = SpoolDirUtil.getParser(
            fs,
            file,
            conf.dataFormat,
            parserFactory,
            offset,
            conf.dataFormatConfig.wholeFileMaxObjectLen,
            rateLimitElEval,
            rateLimitElVars,
            conf.dataFormatConfig.rateLimit
        );
      }

      Map<String, Object> recordHeaderAttr = generateHeaderAttrs(file);

      for (int i = 0; i < maxBatchSize; i++) {
        try {
          Record record;

          try {
            record = parser.parse();
          } catch(RecoverableDataParserException ex) {
            // Propagate partially parsed record to error stream
            record = ex.getUnparsedRecord();
            recordHeaderAttr.put(HeaderAttributeConstants.OFFSET, offset);
            setHeaders(record, recordHeaderAttr);

            errorRecordHandler.onError(new OnRecordErrorException(record, ex.getErrorCode(), ex.getParams()));
            perFileErrorCount++;
            noMoreDataErrorCount++;
            // We'll simply continue reading once this
            continue;
          }

          if (record != null) {
            recordHeaderAttr.put(HeaderAttributeConstants.OFFSET, offset);
            setHeaders(record, recordHeaderAttr);
            batchMaker.addRecord(record);
            offset = parser.getOffset();
            if (offset == null) {
              offset = "0";
            }
            noMoreDataRecordCount++;
            perFileRecordCount++;

          } else {
            parser.close();
            parser = null;
            offset = MINUS_ONE;
            break;
          }
        } catch (ObjectLengthException ex) {
          String exOffset = offset;
          offset = (parser != null) ? parser.getOffset() : MINUS_ONE;
          if (offset == null) {
            offset = "0";
          }
          errorRecordHandler.onError(Errors.SPOOLDIR_02, sourceFile, exOffset, ex);
          perFileErrorCount++;
          noMoreDataErrorCount++;
        }
      }
    } catch (IOException |DataParserException ex) {
      if (ex instanceof ClosedByInterruptException || ex.getCause() instanceof ClosedByInterruptException) {
        //If the pipeline was stopped, we may get a ClosedByInterruptException while reading avro data.
        //This is because the thread is interrupted when the pipeline is stopped.
        //Instead of sending the file to error, publish batch and move one.
      } else {
        offset = MINUS_ONE;
        String exOffset;
        if (ex instanceof OverrunException) {
          exOffset = String.valueOf(((OverrunException) ex).getStreamOffset());
        } else {
          try {
            exOffset = (parser != null) ? parser.getOffset() : MINUS_ONE;
          } catch (IOException ex1) {
            LOG.warn("Could not get the file offset to report with error, reason: {}", ex1.toString(), ex);
            exOffset = MINUS_ONE;
          }
        }
        switch (context.getOnErrorRecord()) {
          case DISCARD:
            break;
          case TO_ERROR:
            // we failed to produce a record, which leaves the input file in an unknown state. all we can do here is
            // throw an exception.
            throw new BadSpoolFileException(file.getAbsolutePath(), exOffset, ex);
          case STOP_PIPELINE:
            context.reportError(Errors.SPOOLDIR_04, sourceFile, exOffset, ex.toString(), ex);
            throw new StageException(Errors.SPOOLDIR_04, sourceFile, exOffset, ex.toString());
          default:
            throw new IllegalStateException(Utils.format("Unknown OnError value '{}'",
                context.getOnErrorRecord(), ex));
        }
      }
    } finally {
      if (MINUS_ONE.equals(offset)) {
        if (parser != null) {
          try {
            parser.close();
            parser = null;
          } catch (IOException ex) {
            //NOP
          }
        }
      }
    }
    return offset;
  }
public String generateBatch(WrappedFile file, String offset, int maxBatchSize, BatchMaker batchMaker) throws
      StageException,
      BadSpoolFileException {
    if (offset == null) {
      offset = "0";
    }
    String sourceFile = file.getFileName();
    try {
      if (parser == null) {
        parser = SpoolDirUtil.getParser(
            fs,
            file,
            conf.dataFormat,
            parserFactory,
            offset,
            conf.dataFormatConfig.wholeFileMaxObjectLen,
            rateLimitElEval,
            rateLimitElVars,
            conf.dataFormatConfig.rateLimit
        );
      }

      Map<String, Object> recordHeaderAttr = generateHeaderAttrs(file);

      for (int i = 0; i < maxBatchSize; i++) {
        try {
          Record record;

          try {
            record = parser.parse();
          } catch(RecoverableDataParserException ex) {
            // Propagate partially parsed record to error stream
            record = ex.getUnparsedRecord();
            recordHeaderAttr.put(HeaderAttributeConstants.OFFSET, offset);
            setHeaders(record, recordHeaderAttr);

            errorRecordHandler.onError(new OnRecordErrorException(record, ex.getErrorCode(), ex.getParams()));
            perFileErrorCount++;
            noMoreDataErrorCount++;
            // We'll simply continue reading once this
            continue;
          }

          if (record != null) {
            recordHeaderAttr.put(HeaderAttributeConstants.OFFSET, offset);
            setHeaders(record, recordHeaderAttr);
            batchMaker.addRecord(record);
            offset = parser.getOffset();
            if (offset == null) {
              offset = "0";
            }
            noMoreDataRecordCount++;
            perFileRecordCount++;

          } else {
            parser.close();
            parser = null;
            offset = MINUS_ONE;
            break;
          }
        } catch (ObjectLengthException ex) {
          String exOffset = offset;
          offset = (parser != null) ? parser.getOffset() : MINUS_ONE;
          if (offset == null) {
            offset = "0";
          }
          errorRecordHandler.onError(Errors.SPOOLDIR_02, sourceFile, exOffset, ex);
          perFileErrorCount++;
          noMoreDataErrorCount++;
        }
      }
    } catch (IOException |DataParserException ex) {
      if (ex instanceof ClosedByInterruptException || ex.getCause() instanceof ClosedByInterruptException) {
        //If the pipeline was stopped, we may get a ClosedByInterruptException while reading avro data.
        //This is because the thread is interrupted when the pipeline is stopped.
        //Instead of sending the file to error, publish batch and move one.
      } else {
        offset = MINUS_ONE;
        String exOffset;
        if (ex instanceof OverrunException) {
          exOffset = String.valueOf(((OverrunException) ex).getStreamOffset());
        } else {
          try {
            exOffset = (parser != null) ? parser.getOffset() : MINUS_ONE;
          } catch (IOException ex1) {
            LOG.warn("Could not get the file offset to report with error, reason: {}", ex1.toString(), ex);
            exOffset = MINUS_ONE;
          }
        }
        switch (context.getOnErrorRecord()) {
          case DISCARD:
            break;
          case TO_ERROR:
            // we failed to produce a record, which leaves the input file in an unknown state. all we can do here is
            // throw an exception.
            throw new BadSpoolFileException(file.getAbsolutePath(), exOffset, ex);
          case STOP_PIPELINE:
            context.reportError(Errors.SPOOLDIR_04, sourceFile, exOffset, ex.toString(), ex);
            throw new StageException(Errors.SPOOLDIR_04, sourceFile, exOffset, ex.toString());
          default:
            throw new IllegalStateException(Utils.format("Unknown OnError value '{}'",
                context.getOnErrorRecord(), ex));
        }
      }
    } finally {
      if (MINUS_ONE.equals(offset)) {
        if (parser != null) {
          try {
            parser.close();
            parser = null;
          } catch (IOException ex) {
            //NOP
          }
        }
      }
    }
    return offset;
  }
public String generateBatch ( WrappedFile file , String offset , int maxBatchSize , BatchMaker batchMaker ) throws StageException , BadSpoolFileException { if ( offset == null ) { offset = "0" ; } String sourceFile = file . getFileName ( ) ; try { if ( parser == null ) { parser = SpoolDirUtil . getParser ( fs , file , conf . dataFormat , parserFactory , offset , conf . dataFormatConfig . wholeFileMaxObjectLen , rateLimitElEval , rateLimitElVars , conf . dataFormatConfig . rateLimit ) ; } Map < String , Object > recordHeaderAttr = generateHeaderAttrs ( file ) ; for ( int i = 0 ; i < maxBatchSize ; i ++ ) { try { Record record ; try { record = parser . parse ( ) ; } catch ( RecoverableDataParserException ex ) { // Propagate partially parsed record to error stream record = ex . getUnparsedRecord ( ) ; recordHeaderAttr . put ( HeaderAttributeConstants . OFFSET , offset ) ; setHeaders ( record , recordHeaderAttr ) ; errorRecordHandler . onError ( new OnRecordErrorException ( record , ex . getErrorCode ( ) , ex . getParams ( ) ) ) ; perFileErrorCount ++ ; noMoreDataErrorCount ++ ; // We'll simply continue reading once this continue ; } if ( record != null ) { recordHeaderAttr . put ( HeaderAttributeConstants . OFFSET , offset ) ; setHeaders ( record , recordHeaderAttr ) ; batchMaker . addRecord ( record ) ; offset = parser . getOffset ( ) ; if ( offset == null ) { offset = "0" ; } noMoreDataRecordCount ++ ; perFileRecordCount ++ ; } else { parser . close ( ) ; parser = null ; offset = MINUS_ONE ; break ; } } catch ( ObjectLengthException ex ) { String exOffset = offset ; offset = ( parser != null ) ? parser . getOffset ( ) : MINUS_ONE ; if ( offset == null ) { offset = "0" ; } errorRecordHandler . onError ( Errors . SPOOLDIR_02 , sourceFile , exOffset , ex ) ; perFileErrorCount ++ ; noMoreDataErrorCount ++ ; } } } catch ( IOException | DataParserException ex ) { if ( ex instanceof ClosedByInterruptException || ex . getCause ( ) instanceof ClosedByInterruptException ) { //If the pipeline was stopped, we may get a ClosedByInterruptException while reading avro data. //This is because the thread is interrupted when the pipeline is stopped. //Instead of sending the file to error, publish batch and move one. } else { offset = MINUS_ONE ; String exOffset ; if ( ex instanceof OverrunException ) { exOffset = String . valueOf ( ( ( OverrunException ) ex ) . getStreamOffset ( ) ) ; } else { try { exOffset = ( parser != null ) ? parser . getOffset ( ) : MINUS_ONE ; } catch ( IOException ex1 ) { LOG . warn ( "Could not get the file offset to report with error, reason: {}" , ex1 . toString ( ) , ex ) ; exOffset = MINUS_ONE ; } } switch ( context . getOnErrorRecord ( ) ) { case DISCARD : break ; case TO_ERROR : // we failed to produce a record, which leaves the input file in an unknown state. all we can do here is // throw an exception. throw new BadSpoolFileException ( file . getAbsolutePath ( ) , exOffset , ex ) ; case STOP_PIPELINE : context . reportError ( Errors . SPOOLDIR_04 , sourceFile , exOffset , ex . toString ( ) , ex ) ; throw new StageException ( Errors . SPOOLDIR_04 , sourceFile , exOffset , ex . toString ( ) ) ; default : throw new IllegalStateException ( Utils . format ( "Unknown OnError value '{}'" , context . getOnErrorRecord ( ) , ex ) ) ; } } } finally { if ( MINUS_ONE . equals ( offset ) ) { if ( parser != null ) { try { parser . close ( ) ; parser = null ; } catch ( IOException ex ) { //NOP } } } } return offset ; }
Processes a batch from the specified file and offset up to a maximum batch size. If the file is fully processed
it must return -1, otherwise it must return the offset to continue from next invocation.
Processes a batch from the specified file and offset up to a maximum batch size . If the file is fully processed it must return - 1 otherwise it must return the offset to continue from next invocation .