<commit_before>package com.salesforce.storm.spout.sideline.coordinator;

import com.google.common.collect.Maps;
import com.salesforce.storm.spout.sideline.kafka.ConsumerState;
import com.salesforce.storm.spout.sideline.kafka.DelegateSidelineSpout;
import com.salesforce.storm.spout.sideline.persistence.PersistenceAdapter;
import com.salesforce.storm.spout.sideline.persistence.SidelinePayload;
import com.salesforce.storm.spout.sideline.trigger.SidelineRequestIdentifier;
import org.apache.kafka.common.TopicPartition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collections;
import java.util.Map;
import java.util.UUID;

/**
 * Attempts to monitor VirtualSpout progress.
 */
public class SpoutPartitionProgressMonitor {
    private static final Logger logger = LoggerFactory.getLogger(SpoutPartitionProgressMonitor.class);

    final PersistenceAdapter persistenceAdapter;
    final Map<TopicPartition, PartitionProgress> mainProgressMap = Maps.newHashMap();

    public SpoutPartitionProgressMonitor(PersistenceAdapter persistenceAdapter) {
        this.persistenceAdapter = persistenceAdapter;
    }

    public void open(Map spoutConfig) {
        persistenceAdapter.open(spoutConfig);
    }

    public void close() {
        persistenceAdapter.close();
    }

    public Map<TopicPartition, PartitionProgress> getStatus(final DelegateSidelineSpout spout) {
        final SidelineRequestIdentifier sidelineIdentifier = getSidelineRequestIdentifier(spout);

        if (sidelineIdentifier == null) {
            // This is the main
            return handleMainVirtualSpout(spout);
        } else {
            // This is a sideline request virtual spout
            return handleSidelineVirtualSpout(spout);
        }
    }

    private SidelineRequestIdentifier getSidelineRequestIdentifier(final DelegateSidelineSpout spout) {
        final String virtualSpoutId = spout.getVirtualSpoutId();

        // Parse out the SidelineRequestId, this is not ideal.
        final String[] bits = virtualSpoutId.split(":");
        if (bits.length != 2) {
            logger.warn("Unable to parse virtualSpoutId: {}", virtualSpoutId);
            return null;
        }

        // Grab 2nd bit.
        final String sidelineRequestIdString = bits[1];

        if (sidelineRequestIdString.equals("main")) {
            return null;
        } else {
            return new SidelineRequestIdentifier(UUID.fromString(sidelineRequestIdString));
        }

    }

    private Map<TopicPartition, PartitionProgress> handleMainVirtualSpout(final DelegateSidelineSpout spout) {
        final ConsumerState currentState = spout.getCurrentState();

        // Max lag is the MAX lag across all partitions
        // This is not ideal...since its not tied to any specific partition.
        // At best you'll get the MAX lag for all of the partitions your instance is consuming from :/
        final Double maxLag = spout.getMaxLag();

        for (TopicPartition topicPartition : currentState.getTopicPartitions()) {
            final PartitionProgress previousProgress = mainProgressMap.get(topicPartition);
            final long currentOffset = currentState.getOffsetForTopicAndPartition(topicPartition);

            // "Calculate" ending offset by adding currentOffset + maxLag
            final long endingOffset = currentOffset + maxLag.longValue();

            if (previousProgress == null) {
                mainProgressMap.put(topicPartition, new PartitionProgress(currentOffset, currentOffset, endingOffset));
                continue;
            }

            // Build new progress
            mainProgressMap.put(topicPartition, new PartitionProgress(previousProgress.getStartingOffset(), currentOffset, endingOffset));
        }

        return Collections.unmodifiableMap(mainProgressMap);
    }

    private Map<TopicPartition, PartitionProgress> handleSidelineVirtualSpout(final DelegateSidelineSpout spout) {
        // Create return map
        Map<TopicPartition, PartitionProgress> progressMap = Maps.newHashMap();

        final String virtualSpoutId = spout.getVirtualSpoutId();
        final SidelineRequestIdentifier sidelineRequestIdentifier = getSidelineRequestIdentifier(spout);
        final ConsumerState currentState = spout.getCurrentState();

        for (TopicPartition topicPartition : currentState.getTopicPartitions()) {
            // Retrieve status
            final SidelinePayload payload = getPersistenceAdapter().retrieveSidelineRequest(
                sidelineRequestIdentifier,
                topicPartition.partition()
            );

            if (payload == null) {
                // Nothing to do?
                logger.error("Could not find SidelineRequest for Id {}", sidelineRequestIdentifier);
                return null;
            }

            // Get the state
            Long currentOffset = getPersistenceAdapter().retrieveConsumerState(virtualSpoutId, topicPartition.partition());
            if (currentOffset == null) {
                logger.error("Could not find Current State for Id {}, assuming consumer has no previous state", virtualSpoutId);
                continue;
            }

            // Make sure no nulls
            boolean hasError = false;
            if (payload.startingOffset == null) {
                logger.warn("No starting state found for {}", topicPartition);<commit_msg>Sideline Adjust log level for SpoutPartitionProgressMonitor<commit_after>package com.salesforce.storm.spout.sideline.coordinator;

import com.google.common.collect.Maps;
import com.salesforce.storm.spout.sideline.kafka.ConsumerState;
import com.salesforce.storm.spout.sideline.kafka.DelegateSidelineSpout;
import com.salesforce.storm.spout.sideline.persistence.PersistenceAdapter;
import com.salesforce.storm.spout.sideline.persistence.SidelinePayload;
import com.salesforce.storm.spout.sideline.trigger.SidelineRequestIdentifier;
import org.apache.kafka.common.TopicPartition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collections;
import java.util.Map;
import java.util.UUID;

/**
 * Attempts to monitor VirtualSpout progress.
 */
public class SpoutPartitionProgressMonitor {
    private static final Logger logger = LoggerFactory.getLogger(SpoutPartitionProgressMonitor.class);

    final PersistenceAdapter persistenceAdapter;
    final Map<TopicPartition, PartitionProgress> mainProgressMap = Maps.newHashMap();

    public SpoutPartitionProgressMonitor(PersistenceAdapter persistenceAdapter) {
        this.persistenceAdapter = persistenceAdapter;
    }

    public void open(Map spoutConfig) {
        persistenceAdapter.open(spoutConfig);
    }

    public void close() {
        persistenceAdapter.close();
    }

    public Map<TopicPartition, PartitionProgress> getStatus(final DelegateSidelineSpout spout) {
        final SidelineRequestIdentifier sidelineIdentifier = getSidelineRequestIdentifier(spout);

        if (sidelineIdentifier == null) {
            // This is the main
            return handleMainVirtualSpout(spout);
        } else {
            // This is a sideline request virtual spout
            return handleSidelineVirtualSpout(spout);
        }
    }

    private SidelineRequestIdentifier getSidelineRequestIdentifier(final DelegateSidelineSpout spout) {
        final String virtualSpoutId = spout.getVirtualSpoutId();

        // Parse out the SidelineRequestId, this is not ideal.
        final String[] bits = virtualSpoutId.split(":");
        if (bits.length != 2) {
            logger.error("Unable to parse virtualSpoutId: {}", virtualSpoutId);
            return null;
        }

        // Grab 2nd bit.
        final String sidelineRequestIdString = bits[1];

        if (sidelineRequestIdString.equals("main")) {
            return null;
        } else {
            return new SidelineRequestIdentifier(UUID.fromString(sidelineRequestIdString));
        }

    }

    private Map<TopicPartition, PartitionProgress> handleMainVirtualSpout(final DelegateSidelineSpout spout) {
        final ConsumerState currentState = spout.getCurrentState();

        // Max lag is the MAX lag across all partitions
        // This is not ideal...since its not tied to any specific partition.
        // At best you'll get the MAX lag for all of the partitions your instance is consuming from :/
        final Double maxLag = spout.getMaxLag();

        for (TopicPartition topicPartition : currentState.getTopicPartitions()) {
            final PartitionProgress previousProgress = mainProgressMap.get(topicPartition);
            final long currentOffset = currentState.getOffsetForTopicAndPartition(topicPartition);

            // "Calculate" ending offset by adding currentOffset + maxLag
            final long endingOffset = currentOffset + maxLag.longValue();

            if (previousProgress == null) {
                mainProgressMap.put(topicPartition, new PartitionProgress(currentOffset, currentOffset, endingOffset));
                continue;
            }

            // Build new progress
            mainProgressMap.put(topicPartition, new PartitionProgress(previousProgress.getStartingOffset(), currentOffset, endingOffset));
        }

        return Collections.unmodifiableMap(mainProgressMap);
    }

    private Map<TopicPartition, PartitionProgress> handleSidelineVirtualSpout(final DelegateSidelineSpout spout) {
        // Create return map
        Map<TopicPartition, PartitionProgress> progressMap = Maps.newHashMap();

        final String virtualSpoutId = spout.getVirtualSpoutId();
        final SidelineRequestIdentifier sidelineRequestIdentifier = getSidelineRequestIdentifier(spout);
        final ConsumerState currentState = spout.getCurrentState();

        for (TopicPartition topicPartition : currentState.getTopicPartitions()) {
            // Retrieve status
            final SidelinePayload payload = getPersistenceAdapter().retrieveSidelineRequest(
                sidelineRequestIdentifier,
                topicPartition.partition()
            );

            if (payload == null) {
                // Nothing to do?
                logger.error("Could not find SidelineRequest for Id {}", sidelineRequestIdentifier);
                return null;
            }

            // Get the state
            Long currentOffset = getPersistenceAdapter().retrieveConsumerState(virtualSpoutId, topicPartition.partition());
            if (currentOffset == null) {
                logger.info("Could not find Current State for Id {}, assuming consumer has no previous state", virtualSpoutId);
                continue;
            }

            // Make sure no nulls
            boolean hasError = false;
            if (payload.startingOffset == null) {
                logger.warn("No starting state found for {}", topicPartition);
{'fraction_non_alphanumeric': 0.07770394979697305, 'fraction_numerical': 0.0009228497600590623, 'mean_word_length': 2.9435953420669576, 'pattern_counts': {'":': 2, '<': 13, '<?xml version=': 0, '>': 13, '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': '424727', 'n_tokens_mistral': 2733, 'n_tokens_neox': 2625, 'n_words': 652}