<commit_before>/*
 * The MIT License
 *
 *   Copyright (c) 2014, benas (<EMAIL>)
 *
 *   Permission is hereby granted, free of charge, to any person obtaining a copy
 *   of this software and associated documentation files (the "Software"), to deal
 *   in the Software without restriction, including without limitation the rights
 *   to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 *   copies of the Software, and to permit persons to whom the Software is
 *   furnished to do so, subject to the following conditions:
 *
 *   The above copyright notice and this permission notice shall be included in
 *   all copies or substantial portions of the Software.
 *
 *   THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 *   IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 *   FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 *   AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 *   LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 *   OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 *   THE SOFTWARE.
 */

package io.github.benas.easybatch.core.impl;

import io.github.benas.easybatch.core.api.*;
import io.github.benas.easybatch.core.jmx.EasyBatchMonitor;

import javax.management.MBeanServer;
import javax.management.ObjectName;
import java.lang.management.ManagementFactory;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * Core Easy Batch engine implementation.
 *
 * @author benas (<EMAIL>)
 */
public final class EasyBatchEngine implements Callable<EasyBatchReport> {

    private static final Logger LOGGER = Logger.getLogger(EasyBatchEngine.class.getName());

    private RecordReader recordReader;

    private RecordFilter recordFilter;

    private RecordMapper recordMapper;

    private RecordValidator recordValidator;

    private RecordProcessor recordProcessor;

    private EasyBatchMonitor easyBatchMonitor;

    private EasyBatchReport easyBatchReport;


    EasyBatchEngine(final RecordReader recordReader, final RecordFilter recordFilter, final RecordMapper recordMapper,
                    final RecordValidator recordValidator, final RecordProcessor recordProcessor) {
        this.recordReader = recordReader;
        this.recordFilter = recordFilter;
        this.recordMapper = recordMapper;
        this.recordValidator = recordValidator;
        this.recordProcessor = recordProcessor;
        easyBatchReport = new EasyBatchReport();
        easyBatchMonitor = new EasyBatchMonitor(easyBatchReport);
        configureJmxMBean();
    }

    @Override
    public EasyBatchReport call() {

        LOGGER.info("Initializing easy batch engine");
        try {
            recordReader.open();
        } catch (Exception e) {
            LOGGER.log(Level.SEVERE, "An exception occurred during opening data source reader", e);
            return null;
        }

        String dataSourceName = recordReader.getDataSourceName();
        LOGGER.info("Data source: " + dataSourceName);
        easyBatchReport.setDataSource(dataSourceName);

        Long totalRecords = recordReader.getTotalRecords();
        LOGGER.info("Total records = " + (totalRecords == null ? "N/A" : totalRecords));
        LOGGER.info("easy batch engine is running...");

        easyBatchReport.setTotalRecords(totalRecords);
        easyBatchReport.setStartTime(System.currentTimeMillis());

        long currentRecordNumber;

        while (recordReader.hasNextRecord()) {

            long currentRecordProcessingStartTime = System.currentTimeMillis();

            //read next record
            Record currentRecord;
            try {
                currentRecord = recordReader.readNextRecord();
            } catch (Exception e) {
                LOGGER.log(Level.SEVERE, "An exception occurred during reading next data source record", e);
                return null;
            }
            currentRecordNumber = currentRecord.getNumber();
            easyBatchReport.setCurrentRecordNumber(currentRecordNumber);

            //filter record if any
            if (recordFilter.filterRecord(currentRecord)) {
                LOGGER.log(Level.INFO, "Record #" + currentRecordNumber + " [" + currentRecord + "] has been filtered");
                easyBatchReport.addFilteredRecord(currentRecordNumber);
                easyBatchReport.addProcessingTime(currentRecordNumber, System.currentTimeMillis() - currentRecordProcessingStartTime);
                continue;
            }

            //map record
            Object typedRecord;
            try {
                typedRecord = recordMapper.mapRecord(currentRecord);
            } catch (Exception e) {
                LOGGER.log(Level.SEVERE, "Error while mapping record #" + currentRecordNumber + " [" + currentRecord + "] : " + e.getMessage(), e);
                easyBatchReport.addIgnoredRecord(currentRecordNumber);
                easyBatchReport.addProcessingTime(currentRecordNumber, System.currentTimeMillis() - currentRecordProcessingStartTime);
                continue;
            }

            //validate record
            Set<ValidationError> validationsErrors = recordValidator.validateRecord(typedRecord);

            if (!validationsErrors.isEmpty()) {
                StringBuilder stringBuilder = new StringBuilder();
                for (ValidationError validationError : validationsErrors) {
                    stringBuilder.append(validationError.getMessage()).append(" | ");
                }
                LOGGER.log(Level.SEVERE, "Error while validating record #" + currentRecordNumber + " [" + currentRecord + "] : " + stringBuilder.toString());
                easyBatchReport.addRejectedRecord(currentRecordNumber);
                easyBatchReport.addProcessingTime(currentRecordNumber, System.currentTimeMillis() - currentRecordProcessingStartTime);
                continue;
            }

            //process record
            try {
                recordProcessor.processRecord(typedRecord);
            } catch (Exception e) {
                LOGGER.log(Level.SEVERE, "Error while processing record #" + currentRecordNumber + "[" + currentRecord + "]", e);
                easyBatchReport.addErrorRecord(currentRecordNumber);
                easyBatchReport.addProcessingTime(currentRecordNumber, System.currentTimeMillis() - currentRecordProcessingStartTime);
            }

            //log processing time for the current record
            easyBatchReport.addProcessingTime(currentRecordNumber, System.currentTimeMillis() - currentRecordProcessingStartTime);

        }

        easyBatchReport.setEndTime(System.currentTimeMillis());
        easyBatchReport.setEasyBatchResult(recordProcessor.getEasyBatchResult());
<commit_msg>fix issue#16 : Easy Batch crashes if an unexpected exception occurs during data validation<commit_after>/*
 * The MIT License
 *
 *   Copyright (c) 2014, benas (<EMAIL>)
 *
 *   Permission is hereby granted, free of charge, to any person obtaining a copy
 *   of this software and associated documentation files (the "Software"), to deal
 *   in the Software without restriction, including without limitation the rights
 *   to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 *   copies of the Software, and to permit persons to whom the Software is
 *   furnished to do so, subject to the following conditions:
 *
 *   The above copyright notice and this permission notice shall be included in
 *   all copies or substantial portions of the Software.
 *
 *   THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 *   IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 *   FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 *   AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 *   LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 *   OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 *   THE SOFTWARE.
 */

package io.github.benas.easybatch.core.impl;

import io.github.benas.easybatch.core.api.*;
import io.github.benas.easybatch.core.jmx.EasyBatchMonitor;

import javax.management.MBeanServer;
import javax.management.ObjectName;
import java.lang.management.ManagementFactory;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * Core Easy Batch engine implementation.
 *
 * @author benas (<EMAIL>)
 */
public final class EasyBatchEngine implements Callable<EasyBatchReport> {

    private static final Logger LOGGER = Logger.getLogger(EasyBatchEngine.class.getName());

    private RecordReader recordReader;

    private RecordFilter recordFilter;

    private RecordMapper recordMapper;

    private RecordValidator recordValidator;

    private RecordProcessor recordProcessor;

    private EasyBatchMonitor easyBatchMonitor;

    private EasyBatchReport easyBatchReport;


    EasyBatchEngine(final RecordReader recordReader, final RecordFilter recordFilter, final RecordMapper recordMapper,
                    final RecordValidator recordValidator, final RecordProcessor recordProcessor) {
        this.recordReader = recordReader;
        this.recordFilter = recordFilter;
        this.recordMapper = recordMapper;
        this.recordValidator = recordValidator;
        this.recordProcessor = recordProcessor;
        easyBatchReport = new EasyBatchReport();
        easyBatchMonitor = new EasyBatchMonitor(easyBatchReport);
        configureJmxMBean();
    }

    @Override
    public EasyBatchReport call() {

        LOGGER.info("Initializing easy batch engine");
        try {
            recordReader.open();
        } catch (Exception e) {
            LOGGER.log(Level.SEVERE, "An exception occurred during opening data source reader", e);
            return null;
        }

        String dataSourceName = recordReader.getDataSourceName();
        LOGGER.info("Data source: " + dataSourceName);
        easyBatchReport.setDataSource(dataSourceName);

        Long totalRecords = recordReader.getTotalRecords();
        LOGGER.info("Total records = " + (totalRecords == null ? "N/A" : totalRecords));
        LOGGER.info("easy batch engine is running...");

        easyBatchReport.setTotalRecords(totalRecords);
        easyBatchReport.setStartTime(System.currentTimeMillis());

        long currentRecordNumber;

        while (recordReader.hasNextRecord()) {

            long currentRecordProcessingStartTime = System.currentTimeMillis();

            //read next record
            Record currentRecord;
            try {
                currentRecord = recordReader.readNextRecord();
            } catch (Exception e) {
                LOGGER.log(Level.SEVERE, "An exception occurred during reading next data source record", e);
                return null;
            }
            currentRecordNumber = currentRecord.getNumber();
            easyBatchReport.setCurrentRecordNumber(currentRecordNumber);

            //filter record if any
            if (recordFilter.filterRecord(currentRecord)) {
                LOGGER.log(Level.INFO, "Record #" + currentRecordNumber + " [" + currentRecord + "] has been filtered");
                easyBatchReport.addFilteredRecord(currentRecordNumber);
                easyBatchReport.addProcessingTime(currentRecordNumber, System.currentTimeMillis() - currentRecordProcessingStartTime);
                continue;
            }

            //map record
            Object typedRecord;
            try {
                typedRecord = recordMapper.mapRecord(currentRecord);
            } catch (Exception e) {
                LOGGER.log(Level.SEVERE, "Error while mapping record #" + currentRecordNumber + " [" + currentRecord + "] : " + e.getMessage(), e);
                easyBatchReport.addIgnoredRecord(currentRecordNumber);
                easyBatchReport.addProcessingTime(currentRecordNumber, System.currentTimeMillis() - currentRecordProcessingStartTime);
                continue;
            }

            //validate record
            try {
                Set<ValidationError> validationsErrors = recordValidator.validateRecord(typedRecord);

                if (!validationsErrors.isEmpty()) {
                    StringBuilder stringBuilder = new StringBuilder();
                    for (ValidationError validationError : validationsErrors) {
                        stringBuilder.append(validationError.getMessage()).append(" | ");
                    }
                    LOGGER.log(Level.SEVERE, "Record #" + currentRecordNumber + " [" + currentRecord + "] is not valid : " + stringBuilder.toString());
                    easyBatchReport.addRejectedRecord(currentRecordNumber);
                    easyBatchReport.addProcessingTime(currentRecordNumber, System.currentTimeMillis() - currentRecordProcessingStartTime);
                    continue;
                }
            } catch(Exception e) {
                LOGGER.log(Level.SEVERE, "An exception occurred while validating record #" + currentRecordNumber + " [" + currentRecord + "]", e);
                easyBatchReport.addRejectedRecord(currentRecordNumber);
                easyBatchReport.addProcessingTime(currentRecordNumber, System.currentTimeMillis() - currentRecordProcessingStartTime);
                continue;
            }

            //process record
            try {
                recordProcessor.processRecord(typedRecord);
            } catch (Exception e) {
                LOGGER.log(Level.SEVERE, "Error while processing record #" + currentRecordNumber + "[" + currentRecord + "]", e);
                easyBatchReport.addErrorRecord(currentRecordNumber);
                easyBatchReport.addProcessingTime(currentRecordNumber, System.currentTimeMillis() - currentRecordProcessingStartTime);
            }

            //log processing time for the current record
            easyBatchReport.addProcessingTime(currentRecordNumber, System.currentTimeMillis() - currentRecordProcessingStartTime);

        }

        easyBatchReport.setEndTime(System.currentTimeMillis());
        easyBatchReport.setEasyBatchResult(recordProcessor.getEasyBatchResult());

{'fraction_non_alphanumeric': 0.08049492562213263, 'fraction_numerical': 0.0006951202558042541, 'mean_word_length': 2.803066349458102, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1797369', 'n_tokens_mistral': 3462, 'n_tokens_neox': 3334, 'n_words': 926}