<commit_before>/*
 * Copyright 2016 Crown Copyright
 *
 * Licensed 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 stroom.index.server;

import javax.annotation.Resource;

import stroom.pipeline.server.errorhandler.ProcessException;
import stroom.util.spring.StroomScope;
import org.apache.lucene.document.Document;
import org.apache.lucene.document.Field;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;
import org.xml.sax.Attributes;
import org.xml.sax.Locator;
import org.xml.sax.SAXException;

import stroom.index.server.CachedIndexService.CachedIndex;
import stroom.index.shared.Index;
import stroom.index.shared.IndexShardKey;
import stroom.query.shared.IndexField;
import stroom.query.shared.IndexFieldType;
import stroom.query.shared.IndexFieldsMap;
import stroom.pipeline.server.LocationFactoryProxy;
import stroom.pipeline.server.errorhandler.ErrorReceiverProxy;
import stroom.pipeline.server.errorhandler.ErrorStatistics;
import stroom.pipeline.server.errorhandler.LoggedException;
import stroom.pipeline.server.factory.ConfigurableElement;
import stroom.pipeline.server.factory.ElementIcons;
import stroom.pipeline.server.factory.PipelineProperty;
import stroom.pipeline.server.filter.AbstractXMLFilter;
import stroom.pipeline.shared.data.PipelineElementType;
import stroom.pipeline.shared.data.PipelineElementType.Category;
import stroom.pipeline.state.StreamHolder;
import stroom.util.CharBuffer;
import stroom.util.date.DateUtil;
import stroom.util.logging.StroomLogger;
import stroom.util.shared.Severity;

import java.util.concurrent.locks.Lock;

/**
 * The index filter... takes the index XML and builds the LUCENE documents
 */
@Component
@Scope(StroomScope.PROTOTYPE)
@ConfigurableElement(type = "IndexingFilter", category = Category.FILTER, roles = { PipelineElementType.ROLE_TARGET,
        PipelineElementType.ROLE_HAS_TARGETS, PipelineElementType.VISABILITY_SIMPLE }, icon = ElementIcons.INDEX)
public class IndexingFilter extends AbstractXMLFilter {
    private static final StroomLogger LOGGER = StroomLogger.getLogger(IndexingFilter.class);

    private static final StripedLock WRITER_KEY_LOCKS = new StripedLock();

    private static final String RECORD = "record";
    private static final String DATA = "data";
    private static final String NAME = "name";
    private static final String VALUE = "value";

    @Resource
    private StreamHolder streamHolder;
    @Resource
    private LocationFactoryProxy locationFactory;
    @Resource
    private IndexShardWriterCache indexShardCache;
    @Resource
    private ErrorReceiverProxy errorReceiverProxy;
    @Resource
    private CachedIndexService cachedIndexService;

    private IndexFieldsMap indexFieldsMap;

    private Index index;
    private IndexShardKey indexShardKey;
    private IndexShardWriter indexShardWriter;

    private final CharBuffer debugBuffer = new CharBuffer(10);

    private Document document;

    private int fieldsIndexed = 0;

    private Locator locator;

    /**
     * Initialise
     */
    @Override
    public void startProcessing() {
        try {
            if (index == null) {
                log(Severity.FATAL_ERROR, "Index has not been set", null);
                throw new LoggedException("Index has not been set");
            }

            // Get the index and index fields from the cache.
            final CachedIndex cachedIndex = cachedIndexService.get(index);
            if (cachedIndex == null) {
                log(Severity.FATAL_ERROR, "Unable to load index", null);
                throw new LoggedException("Unable to load index");
            }

            index = cachedIndex.getIndex();
            indexFieldsMap = cachedIndex.getIndexFieldsMap();

            // Create a key to create shards with.
            if (streamHolder == null || streamHolder.getStream() == null) {
                // Many tests don't use streams so where this is the case just
                // create a basic key.
                indexShardKey = IndexShardKeyUtil.createTestKey(index);
            } else {
                final long timeMs = streamHolder.getStream().getCreateMs();
                indexShardKey = IndexShardKeyUtil.createTimeBasedPartition(index, timeMs);
            }
        } finally {
            super.startProcessing();
        }
    }

    /**
     * Sets the locator to use when reporting errors.
     *
     * @param locator
     *            The locator to use.
     */
    @Override
    public void setDocumentLocator(final Locator locator) {
        this.locator = locator;
        super.setDocumentLocator(locator);
    }

    @Override
    public void startElement(final String uri, final String localName, final String qName, final Attributes atts)
            throws SAXException {
        if (DATA.equals(localName) && document != null) {
            String name = atts.getValue(NAME);
            String value = atts.getValue(VALUE);
            if (name != null && value != null) {
                name = name.trim();
                value = value.trim();

                if (name.length() > 0 && value.length() > 0) {
                    // See if we can get this field.
                    final IndexField indexField = indexFieldsMap.get(name);
                    if (indexField != null) {
                        // Index the current content if we are to store or index
                        // this field.
                        if (indexField.isIndexed() || indexField.isStored()) {
                            processIndexContent(indexField, value);
                        }
                    } else {
                        log(Severity.WARNING, "Attempt to index unknown field: " + name, null);
                    }
                }
            }
        } else if (RECORD.equals(localName)) {
            // Create a document to store fields in.
            document = new Document();
        }

        super.startElement(uri, localName, qName, atts);
    }

    @Override
    public void endElement(final String uri, final String localName, final String qName) throws SAXException {
        if (RECORD.equals(localName)) {
            processDocument();
            document = null;

            // Reset the count of how many fields we have indexed for the
            // current event.
            fieldsIndexed = 0;

            if (errorReceiverProxy.getErrorReceiver() != null
                    && errorReceiverProxy.getErrorReceiver() instanceof ErrorStatistics) {
                ((ErrorStatistics) errorReceiverProxy.getErrorReceiver()).checkRecord(-1);
            }
        }

        super.endElement(uri, localName, qName);
    }

    private void processDocument() throws SAXException {
        // Write the document if we have dropped out of the record element and
        // have indexed some fields.
        if (fieldsIndexed > 0) {
            try {
                boolean success = false;
                while (!success) {
                    success = indexShardWriter != null && indexShardWriter.addDocument(document);

                    if (!success) {
                        // If we failed then try under lock to make sure we get a new writer.
                        final Lock lock = WRITER_KEY_LOCKS.getLockForKey(indexShardKey);
                        lock.lock();
                        try {
                            // Ask the cache for the current one (it might have been changed by another thread) and try again.
                            indexShardWriter = getIndexShardWriter();
                            success = indexShardWriter.addDocument(document);

                            if (!success) {
                                // Failed to add it so remove this object from the cache and try to get another one.
                                indexShardCache.remove(indexShardKey);

                                // Ask the pool for another one and try again
                                final IndexShardWriter newWriter = getIndexShardWriter();
                                if (newWriter == indexShardWriter) {
                                    LOGGER.fatal("Expected a new writer but got the same one back!!!");
                                    throw new IndexException("Expected a new writer but got the same one back!!!");
                                }

                                indexShardWriter = newWriter;
                            }

                        } finally {
                            lock.unlock();
                        }
                    }
                }
            } catch (final RuntimeException e) {
                log(Severity.FATAL_ERROR, e.getMessage(), e);
                // Terminate processing as this is a fatal error.
                throw new LoggedException(e.getMessage(), e);
            }
        }
    }
<commit_msg>Issue #220 : Changed 'Expected a new writer...' fatal error to warning as the condition in question might be acceptable.<commit_after>/*
 * Copyright 2016 Crown Copyright
 *
 * Licensed 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 stroom.index.server;

import javax.annotation.Resource;

import stroom.pipeline.server.errorhandler.ProcessException;
import stroom.util.spring.StroomScope;
import org.apache.lucene.document.Document;
import org.apache.lucene.document.Field;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;
import org.xml.sax.Attributes;
import org.xml.sax.Locator;
import org.xml.sax.SAXException;

import stroom.index.server.CachedIndexService.CachedIndex;
import stroom.index.shared.Index;
import stroom.index.shared.IndexShardKey;
import stroom.query.shared.IndexField;
import stroom.query.shared.IndexFieldType;
import stroom.query.shared.IndexFieldsMap;
import stroom.pipeline.server.LocationFactoryProxy;
import stroom.pipeline.server.errorhandler.ErrorReceiverProxy;
import stroom.pipeline.server.errorhandler.ErrorStatistics;
import stroom.pipeline.server.errorhandler.LoggedException;
import stroom.pipeline.server.factory.ConfigurableElement;
import stroom.pipeline.server.factory.ElementIcons;
import stroom.pipeline.server.factory.PipelineProperty;
import stroom.pipeline.server.filter.AbstractXMLFilter;
import stroom.pipeline.shared.data.PipelineElementType;
import stroom.pipeline.shared.data.PipelineElementType.Category;
import stroom.pipeline.state.StreamHolder;
import stroom.util.CharBuffer;
import stroom.util.date.DateUtil;
import stroom.util.logging.StroomLogger;
import stroom.util.shared.Severity;

import java.util.concurrent.locks.Lock;

/**
 * The index filter... takes the index XML and builds the LUCENE documents
 */
@Component
@Scope(StroomScope.PROTOTYPE)
@ConfigurableElement(type = "IndexingFilter", category = Category.FILTER, roles = { PipelineElementType.ROLE_TARGET,
        PipelineElementType.ROLE_HAS_TARGETS, PipelineElementType.VISABILITY_SIMPLE }, icon = ElementIcons.INDEX)
public class IndexingFilter extends AbstractXMLFilter {
    private static final StroomLogger LOGGER = StroomLogger.getLogger(IndexingFilter.class);

    private static final StripedLock WRITER_KEY_LOCKS = new StripedLock();

    private static final String RECORD = "record";
    private static final String DATA = "data";
    private static final String NAME = "name";
    private static final String VALUE = "value";

    @Resource
    private StreamHolder streamHolder;
    @Resource
    private LocationFactoryProxy locationFactory;
    @Resource
    private IndexShardWriterCache indexShardCache;
    @Resource
    private ErrorReceiverProxy errorReceiverProxy;
    @Resource
    private CachedIndexService cachedIndexService;

    private IndexFieldsMap indexFieldsMap;

    private Index index;
    private IndexShardKey indexShardKey;
    private IndexShardWriter indexShardWriter;

    private final CharBuffer debugBuffer = new CharBuffer(10);

    private Document document;

    private int fieldsIndexed = 0;

    private Locator locator;

    /**
     * Initialise
     */
    @Override
    public void startProcessing() {
        try {
            if (index == null) {
                log(Severity.FATAL_ERROR, "Index has not been set", null);
                throw new LoggedException("Index has not been set");
            }

            // Get the index and index fields from the cache.
            final CachedIndex cachedIndex = cachedIndexService.get(index);
            if (cachedIndex == null) {
                log(Severity.FATAL_ERROR, "Unable to load index", null);
                throw new LoggedException("Unable to load index");
            }

            index = cachedIndex.getIndex();
            indexFieldsMap = cachedIndex.getIndexFieldsMap();

            // Create a key to create shards with.
            if (streamHolder == null || streamHolder.getStream() == null) {
                // Many tests don't use streams so where this is the case just
                // create a basic key.
                indexShardKey = IndexShardKeyUtil.createTestKey(index);
            } else {
                final long timeMs = streamHolder.getStream().getCreateMs();
                indexShardKey = IndexShardKeyUtil.createTimeBasedPartition(index, timeMs);
            }
        } finally {
            super.startProcessing();
        }
    }

    /**
     * Sets the locator to use when reporting errors.
     *
     * @param locator
     *            The locator to use.
     */
    @Override
    public void setDocumentLocator(final Locator locator) {
        this.locator = locator;
        super.setDocumentLocator(locator);
    }

    @Override
    public void startElement(final String uri, final String localName, final String qName, final Attributes atts)
            throws SAXException {
        if (DATA.equals(localName) && document != null) {
            String name = atts.getValue(NAME);
            String value = atts.getValue(VALUE);
            if (name != null && value != null) {
                name = name.trim();
                value = value.trim();

                if (name.length() > 0 && value.length() > 0) {
                    // See if we can get this field.
                    final IndexField indexField = indexFieldsMap.get(name);
                    if (indexField != null) {
                        // Index the current content if we are to store or index
                        // this field.
                        if (indexField.isIndexed() || indexField.isStored()) {
                            processIndexContent(indexField, value);
                        }
                    } else {
                        log(Severity.WARNING, "Attempt to index unknown field: " + name, null);
                    }
                }
            }
        } else if (RECORD.equals(localName)) {
            // Create a document to store fields in.
            document = new Document();
        }

        super.startElement(uri, localName, qName, atts);
    }

    @Override
    public void endElement(final String uri, final String localName, final String qName) throws SAXException {
        if (RECORD.equals(localName)) {
            processDocument();
            document = null;

            // Reset the count of how many fields we have indexed for the
            // current event.
            fieldsIndexed = 0;

            if (errorReceiverProxy.getErrorReceiver() != null
                    && errorReceiverProxy.getErrorReceiver() instanceof ErrorStatistics) {
                ((ErrorStatistics) errorReceiverProxy.getErrorReceiver()).checkRecord(-1);
            }
        }

        super.endElement(uri, localName, qName);
    }

    private void processDocument() throws SAXException {
        // Write the document if we have dropped out of the record element and
        // have indexed some fields.
        if (fieldsIndexed > 0) {
            try {
                boolean success = false;
                while (!success) {
                    success = indexShardWriter != null && indexShardWriter.addDocument(document);

                    if (!success) {
                        // If we failed then try under lock to make sure we get a new writer.
                        final Lock lock = WRITER_KEY_LOCKS.getLockForKey(indexShardKey);
                        lock.lock();
                        try {
                            // Ask the cache for the current one (it might have been changed by another thread) and try again.
                            indexShardWriter = getIndexShardWriter();
                            success = indexShardWriter.addDocument(document);

                            if (!success) {
                                // Failed to add it so remove this object from the cache and try to get another one.
                                indexShardCache.remove(indexShardKey);

                                // Ask the pool for another one and try again
                                final IndexShardWriter newWriter = getIndexShardWriter();
                                if (newWriter == indexShardWriter) {
                                    LOGGER.warn("Expected a new writer but got the same one back!!!");
//                                    throw new IndexException("Expected a new writer but got the same one back!!!");
                                }

                                indexShardWriter = newWriter;
                            }

                        } finally {
                            lock.unlock();
                        }
                    }
                }
            } catch (final RuntimeException e) {
                log(Severity.FATAL_ERROR, e.getMessage(), e);
                // Terminate processing as this is a fatal error.
                throw new LoggedException(e.getMessage(), e);
            }
        }
    }

{'fraction_non_alphanumeric': 0.0788666878105508, 'fraction_numerical': 0.001850089861507559, 'mean_word_length': 2.0455569864777847, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 6}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6816491', 'n_tokens_mistral': 4735, 'n_tokens_neox': 4556, 'n_words': 1392}