<commit_before>package com.thinkbiganalytics.nifi.v2.core.metadata;

import com.fasterxml.jackson.core.type.TypeReference;

/*-
 * #%L
 * thinkbig-nifi-core-service
 * %%
 * Copyright (C) 2017 ThinkBig Analytics
 * %%
 * 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.
 * #L%
 */

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectReader;
import com.fasterxml.jackson.databind.ObjectWriter;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.thinkbiganalytics.metadata.rest.client.MetadataClient;
import com.thinkbiganalytics.metadata.rest.model.feed.InitializationStatus;
import com.thinkbiganalytics.nifi.core.api.metadata.ActiveWaterMarksCancelledException;
import com.thinkbiganalytics.metadata.rest.model.feed.reindex.FeedDataHistoryReindexParams;
import com.thinkbiganalytics.metadata.rest.model.feed.reindex.HistoryReindexingStatus;
import com.thinkbiganalytics.nifi.core.api.metadata.MetadataRecorder;
import com.thinkbiganalytics.nifi.core.api.metadata.WaterMarkActiveException;

import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.exception.ProcessException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.URI;
import java.util.HashSet;
import java.util.Map;
import java.util.NavigableMap;
import java.util.Optional;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.ExecutionException;

import javax.annotation.Nonnull;

public class MetadataClientRecorder implements MetadataRecorder {


    private static final Logger log = LoggerFactory.getLogger(MetadataClientRecorder.class);

//    private static final int FEED_INIT_STATUS_EXPIRE_SEC = 60;
    private static final int FEED_INIT_STATUS_CACHE_SIZE = 1024;
    
    private static final String CURRENT_WATER_MARKS_ATTR = "activeWaterMarks";
    private static final TypeReference<NavigableMap<String, WaterMarkParam>> WM_MAP_TYPE = new TypeReference<NavigableMap<String, WaterMarkParam>>() { };
    private static final ObjectReader WATER_MARKS_READER = new ObjectMapper().reader().forType(WM_MAP_TYPE);
    private static final ObjectWriter WATER_MARKS_WRITER = new ObjectMapper().writer().forType(WM_MAP_TYPE);

    private MetadataClient client;
    private NavigableMap<String, Long> activeWaterMarks = new ConcurrentSkipListMap<>();
    
    private volatile Cache<String, Optional<InitializationStatus>> initStatusCache;

    /**
     * constructor creates a MetadataClientRecorder with the default URI constant
     */
    public MetadataClientRecorder() {
        this(URI.create("http://localhost:8420/api/v1/metadata"));
    }

    /**
     * constructor creates a MetadataClientRecorder with the URI provided
     *
     * @param baseUri the REST endpoint of the Metadata recorder
     */
    public MetadataClientRecorder(URI baseUri) {
        this(new MetadataClient(baseUri));
    }

    /**
     * constructor creates a MetadataClientProvider with the required {@link MetadataClientRecorder}
     *
     * @param client the MetadataClient will be used to connect with the Metadata store
     */
    public MetadataClientRecorder(MetadataClient client) {
        this.client = client;
        getInitStatusCache();
    }
    
    /* (non-Javadoc)
     * @see com.thinkbiganalytics.nifi.core.api.metadata.MetadataRecorder#loadWaterMark(org.apache.nifi.processor.ProcessSession, org.apache.nifi.flowfile.FlowFile, java.lang.String, java.lang.String)
     */
    @Override
    public FlowFile loadWaterMark(ProcessSession session, FlowFile ff, String feedId, String waterMarkName, String parameterName, String initialValue) throws WaterMarkActiveException {
        long timestamp = recordActiveWaterMark(feedId, waterMarkName);

        String value = getHighWaterMarkValue(feedId, waterMarkName).orElse(initialValue);
        FlowFile resultFF = addToCurrentWaterMarksAttr(session, ff, waterMarkName, parameterName, timestamp);
        resultFF = session.putAttribute(resultFF, parameterName, value);
        return session.putAttribute(resultFF, originalValueParameterName(parameterName), value);
    }

    /* (non-Javadoc)
     * @see com.thinkbiganalytics.nifi.core.api.metadata.MetadataRecorder#cancelWaterMark(org.apache.nifi.processor.ProcessSession, org.apache.nifi.flowfile.FlowFile, java.lang.String, java.lang.String)
     */
    @Override
    public boolean cancelWaterMark(String feedId, String waterMarkName) {
        return cancelActiveWaterMark(feedId, waterMarkName);
    }

    /* (non-Javadoc)
     * @see com.thinkbiganalytics.nifi.core.api.metadata.MetadataRecorder#cancelAndLoadWaterMark(org.apache.nifi.processor.ProcessSession, org.apache.nifi.flowfile.FlowFile, java.lang.String, java.lang.String, java.lang.String, java.lang.String)
     */
    @Override
    public FlowFile cancelAndLoadWaterMark(ProcessSession session, FlowFile ff, String feedId, String waterMarkName, String parameterName, String initialValue) throws WaterMarkActiveException {
        cancelActiveWaterMark(feedId, waterMarkName);
        return loadWaterMark(session, ff, feedId, waterMarkName, parameterName, initialValue);
    }

    /* (non-Javadoc)
     * @see com.thinkbiganalytics.nifi.core.api.metadata.MetadataRecorder#recordWaterMark(org.apache.nifi.processor.ProcessSession, org.apache.nifi.flowfile.FlowFile, java.lang.String, java.lang.String, java.io.Serializable)
     */
    @Override
    public FlowFile recordWaterMark(ProcessSession session, FlowFile ff, String feedId, String waterMarkName, String parameterName, String newValue) {
        Map<String, WaterMarkParam> actives = getCurrentWaterMarksAttr(ff);

        if (actives.containsKey(waterMarkName)) {
            return session.putAttribute(ff, parameterName, newValue);
        } else {
            throw new IllegalStateException("No active high-water mark named \"" + waterMarkName + "\"");
        }<commit_msg>KYLO-2171: Ensures active water mark is cancelled when kylo-services comms are lost during load.<commit_after>package com.thinkbiganalytics.nifi.v2.core.metadata;

import com.fasterxml.jackson.core.type.TypeReference;

/*-
 * #%L
 * thinkbig-nifi-core-service
 * %%
 * Copyright (C) 2017 ThinkBig Analytics
 * %%
 * 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.
 * #L%
 */

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectReader;
import com.fasterxml.jackson.databind.ObjectWriter;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.thinkbiganalytics.metadata.rest.client.MetadataClient;
import com.thinkbiganalytics.metadata.rest.model.feed.InitializationStatus;
import com.thinkbiganalytics.nifi.core.api.metadata.ActiveWaterMarksCancelledException;
import com.thinkbiganalytics.metadata.rest.model.feed.reindex.FeedDataHistoryReindexParams;
import com.thinkbiganalytics.metadata.rest.model.feed.reindex.HistoryReindexingStatus;
import com.thinkbiganalytics.nifi.core.api.metadata.MetadataRecorder;
import com.thinkbiganalytics.nifi.core.api.metadata.WaterMarkActiveException;

import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.exception.ProcessException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.URI;
import java.util.HashSet;
import java.util.Map;
import java.util.NavigableMap;
import java.util.Optional;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.ExecutionException;

import javax.annotation.Nonnull;

public class MetadataClientRecorder implements MetadataRecorder {


    private static final Logger log = LoggerFactory.getLogger(MetadataClientRecorder.class);

//    private static final int FEED_INIT_STATUS_EXPIRE_SEC = 60;
    private static final int FEED_INIT_STATUS_CACHE_SIZE = 1024;
    
    private static final String CURRENT_WATER_MARKS_ATTR = "activeWaterMarks";
    private static final TypeReference<NavigableMap<String, WaterMarkParam>> WM_MAP_TYPE = new TypeReference<NavigableMap<String, WaterMarkParam>>() { };
    private static final ObjectReader WATER_MARKS_READER = new ObjectMapper().reader().forType(WM_MAP_TYPE);
    private static final ObjectWriter WATER_MARKS_WRITER = new ObjectMapper().writer().forType(WM_MAP_TYPE);

    private MetadataClient client;
    private NavigableMap<String, Long> activeWaterMarks = new ConcurrentSkipListMap<>();
    
    private volatile Cache<String, Optional<InitializationStatus>> initStatusCache;

    /**
     * constructor creates a MetadataClientRecorder with the default URI constant
     */
    public MetadataClientRecorder() {
        this(URI.create("http://localhost:8420/api/v1/metadata"));
    }

    /**
     * constructor creates a MetadataClientRecorder with the URI provided
     *
     * @param baseUri the REST endpoint of the Metadata recorder
     */
    public MetadataClientRecorder(URI baseUri) {
        this(new MetadataClient(baseUri));
    }

    /**
     * constructor creates a MetadataClientProvider with the required {@link MetadataClientRecorder}
     *
     * @param client the MetadataClient will be used to connect with the Metadata store
     */
    public MetadataClientRecorder(MetadataClient client) {
        this.client = client;
        getInitStatusCache();
    }
    
    /* (non-Javadoc)
     * @see com.thinkbiganalytics.nifi.core.api.metadata.MetadataRecorder#loadWaterMark(org.apache.nifi.processor.ProcessSession, org.apache.nifi.flowfile.FlowFile, java.lang.String, java.lang.String)
     */
    @Override
    public FlowFile loadWaterMark(ProcessSession session, FlowFile ff, String feedId, String waterMarkName, String parameterName, String initialValue) throws WaterMarkActiveException {
        long timestamp = recordActiveWaterMark(feedId, waterMarkName);

        try {
            String value = getHighWaterMarkValue(feedId, waterMarkName).orElse(initialValue);
            FlowFile resultFF = addToCurrentWaterMarksAttr(session, ff, waterMarkName, parameterName, timestamp);
            resultFF = session.putAttribute(resultFF, parameterName, value);
            return session.putAttribute(resultFF, originalValueParameterName(parameterName), value);
        } catch (Exception e) {
            cancelActiveWaterMark(feedId, waterMarkName);
            throw e;
        }
    }

    /* (non-Javadoc)
     * @see com.thinkbiganalytics.nifi.core.api.metadata.MetadataRecorder#cancelWaterMark(org.apache.nifi.processor.ProcessSession, org.apache.nifi.flowfile.FlowFile, java.lang.String, java.lang.String)
     */
    @Override
    public boolean cancelWaterMark(String feedId, String waterMarkName) {
        return cancelActiveWaterMark(feedId, waterMarkName);
    }

    /* (non-Javadoc)
     * @see com.thinkbiganalytics.nifi.core.api.metadata.MetadataRecorder#cancelAndLoadWaterMark(org.apache.nifi.processor.ProcessSession, org.apache.nifi.flowfile.FlowFile, java.lang.String, java.lang.String, java.lang.String, java.lang.String)
     */
    @Override
    public FlowFile cancelAndLoadWaterMark(ProcessSession session, FlowFile ff, String feedId, String waterMarkName, String parameterName, String initialValue) throws WaterMarkActiveException {
        cancelActiveWaterMark(feedId, waterMarkName);
        return loadWaterMark(session, ff, feedId, waterMarkName, parameterName, initialValue);
    }

    /* (non-Javadoc)
     * @see com.thinkbiganalytics.nifi.core.api.metadata.MetadataRecorder#recordWaterMark(org.apache.nifi.processor.ProcessSession, org.apache.nifi.flowfile.FlowFile, java.lang.String, java.lang.String, java.io.Serializable)
     */
    @Override
    public FlowFile recordWaterMark(ProcessSession session, FlowFile ff, String feedId, String waterMarkName, String parameterName, String newValue) {
        Map<String, WaterMarkParam> actives = getCurrentWaterMarksAttr(ff);

        if (actives.containsKey(waterMarkName)) {
            return session.putAttribute(ff, parameterName, newValue);
        } else {
            throw new IllegalStateException("No active high-water mark named \"" + waterMarkName + "\"");
        }
{'fraction_non_alphanumeric': 0.09135264975228945, 'fraction_numerical': 0.003603062603212731, 'mean_word_length': 4.444626072742134, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 8}, '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': '3235294', 'n_tokens_mistral': 3834, 'n_tokens_neox': 3702, 'n_words': 839}