<commit_before>package edu.uci.ics.asterix.metadata.feeds;

import java.io.IOException;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicLong;
import java.util.logging.Level;
import java.util.logging.Logger;

import edu.uci.ics.asterix.metadata.feeds.FeedRuntime.FeedRuntimeType;
import edu.uci.ics.asterix.metadata.feeds.SuperFeedManager.FeedReportMessageType;
import edu.uci.ics.hyracks.api.comm.IFrameWriter;
import edu.uci.ics.hyracks.api.dataflow.IOperatorNodePushable;
import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAccessor;

public class FeedFrameWriter implements IFrameWriter {

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

    private IFrameWriter writer;

    private IOperatorNodePushable nodePushable;

    private FeedPolicyEnforcer policyEnforcer;

    private FeedConnectionId feedId;

    private LinkedBlockingQueue<Long> statsOutbox;

    private final boolean collectStatistics;

    private List<ByteBuffer> frames = new ArrayList<ByteBuffer>();

    private Mode mode;

    private String nodeId;

    public static final long FLUSH_THRESHOLD_TIME = 5000;

    private FramePushWait framePushWait;

    private FeedRuntimeType feedRuntimeType;

    private int partition;

    private Timer timer;

    private ExecutorService executorService;

    private FrameTupleAccessor fta;

    public enum Mode {
        FORWARD,
        STORE
    }

    public FeedFrameWriter(IFrameWriter writer, IOperatorNodePushable nodePushable, FeedConnectionId feedId,
            FeedPolicyEnforcer policyEnforcer, String nodeId, FeedRuntimeType feedRuntimeType, int partition,
            FrameTupleAccessor fta) {
        this.writer = writer;
        this.mode = Mode.FORWARD;
        this.nodePushable = nodePushable;
        this.feedId = feedId;
        this.policyEnforcer = policyEnforcer;
        this.feedRuntimeType = feedRuntimeType;
        this.partition = partition;
        this.executorService = FeedManager.INSTANCE.getFeedExecutorService(feedId);
        this.collectStatistics = policyEnforcer.getFeedPolicyAccessor().collectStatistics();
        if (collectStatistics) {
            this.statsOutbox = new LinkedBlockingQueue<Long>();
            framePushWait = new FramePushWait(nodePushable, FLUSH_THRESHOLD_TIME, feedId, nodeId, feedRuntimeType,
                    partition, FLUSH_THRESHOLD_TIME);
            Timer timer = new Timer();
            timer.scheduleAtFixedRate(framePushWait, 0, FLUSH_THRESHOLD_TIME);
        }
        this.fta = fta;
    }

    public Mode getMode() {
        return mode;
    }

    public void setMode(Mode newMode) throws HyracksDataException {
        if (this.mode.equals(newMode)) {
            return;
        }
        if (LOGGER.isLoggable(Level.INFO)) {
            LOGGER.info("Switching to :" + newMode + " from " + this.mode);
        }
        switch (newMode) {
            case FORWARD:
                this.mode = newMode;
                break;
            case STORE:
                this.mode = newMode;
                if (LOGGER.isLoggable(Level.INFO)) {
                    LOGGER.info("Beginning to store frames :");
                    LOGGER.info("Frames accumulated till now:" + frames.size());
                }
                break;
        }

    }

    public List<ByteBuffer> getStoredFrames() {
        return frames;
    }

    public void clear() {
        frames.clear();
    }

    @Override
    public void open() throws HyracksDataException {
        writer.open();
    }

    public void reset() {
        framePushWait.reset();
    }

    @Override
    public void nextFrame(ByteBuffer buffer) throws HyracksDataException {
        switch (mode) {
            case FORWARD:
                try {
                    if (collectStatistics) {
                        fta.reset(buffer);
                        framePushWait.notifyStart();
                        writer.nextFrame(buffer);
                        framePushWait.notifyFinish(fta.getTupleCount());
                    } else {
                        writer.nextFrame(buffer);
                    }
                } catch (Exception e) {
                    if (LOGGER.isLoggable(Level.SEVERE)) {
                        LOGGER.severe("Unable to write frame " + " on behalf of " + nodePushable.getDisplayName());
                    }
                }
                if (frames.size() > 0) {
                    for (ByteBuffer buf : frames) {
                        if (LOGGER.isLoggable(Level.WARNING)) {
                            LOGGER.warning("Flusing OLD frame: " + buf + " on behalf of "
                                    + nodePushable.getDisplayName());
                        }
                        writer.nextFrame(buf);
                    }
                }<commit_msg>log stacktrace when feedframewriter fails to write a frame<commit_after>package edu.uci.ics.asterix.metadata.feeds;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicLong;
import java.util.logging.Level;
import java.util.logging.Logger;

import edu.uci.ics.asterix.metadata.feeds.FeedRuntime.FeedRuntimeType;
import edu.uci.ics.asterix.metadata.feeds.SuperFeedManager.FeedReportMessageType;
import edu.uci.ics.hyracks.api.comm.IFrameWriter;
import edu.uci.ics.hyracks.api.dataflow.IOperatorNodePushable;
import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAccessor;

public class FeedFrameWriter implements IFrameWriter {

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

    private IFrameWriter writer;

    private IOperatorNodePushable nodePushable;

    private FeedPolicyEnforcer policyEnforcer;

    private FeedConnectionId feedId;

    private LinkedBlockingQueue<Long> statsOutbox;

    private final boolean collectStatistics;

    private List<ByteBuffer> frames = new ArrayList<ByteBuffer>();

    private Mode mode;

    private String nodeId;

    public static final long FLUSH_THRESHOLD_TIME = 5000;

    private FramePushWait framePushWait;

    private FeedRuntimeType feedRuntimeType;

    private int partition;

    private Timer timer;

    private ExecutorService executorService;

    private FrameTupleAccessor fta;

    public enum Mode {
        FORWARD,
        STORE
    }

    public FeedFrameWriter(IFrameWriter writer, IOperatorNodePushable nodePushable, FeedConnectionId feedId,
            FeedPolicyEnforcer policyEnforcer, String nodeId, FeedRuntimeType feedRuntimeType, int partition,
            FrameTupleAccessor fta) {
        this.writer = writer;
        this.mode = Mode.FORWARD;
        this.nodePushable = nodePushable;
        this.feedId = feedId;
        this.policyEnforcer = policyEnforcer;
        this.feedRuntimeType = feedRuntimeType;
        this.partition = partition;
        this.executorService = FeedManager.INSTANCE.getFeedExecutorService(feedId);
        this.collectStatistics = policyEnforcer.getFeedPolicyAccessor().collectStatistics();
        if (collectStatistics) {
            this.statsOutbox = new LinkedBlockingQueue<Long>();
            framePushWait = new FramePushWait(nodePushable, FLUSH_THRESHOLD_TIME, feedId, nodeId, feedRuntimeType,
                    partition, FLUSH_THRESHOLD_TIME);
            Timer timer = new Timer();
            timer.scheduleAtFixedRate(framePushWait, 0, FLUSH_THRESHOLD_TIME);
        }
        this.fta = fta;
    }

    public Mode getMode() {
        return mode;
    }

    public void setMode(Mode newMode) throws HyracksDataException {
        if (this.mode.equals(newMode)) {
            return;
        }
        if (LOGGER.isLoggable(Level.INFO)) {
            LOGGER.info("Switching to :" + newMode + " from " + this.mode);
        }
        switch (newMode) {
            case FORWARD:
                this.mode = newMode;
                break;
            case STORE:
                this.mode = newMode;
                if (LOGGER.isLoggable(Level.INFO)) {
                    LOGGER.info("Beginning to store frames :");
                    LOGGER.info("Frames accumulated till now:" + frames.size());
                }
                break;
        }

    }

    public List<ByteBuffer> getStoredFrames() {
        return frames;
    }

    public void clear() {
        frames.clear();
    }

    @Override
    public void open() throws HyracksDataException {
        writer.open();
    }

    public void reset() {
        framePushWait.reset();
    }

    @Override
    public void nextFrame(ByteBuffer buffer) throws HyracksDataException {
        switch (mode) {
            case FORWARD:
                try {
                    if (collectStatistics) {
                        fta.reset(buffer);
                        framePushWait.notifyStart();
                        writer.nextFrame(buffer);
                        framePushWait.notifyFinish(fta.getTupleCount());
                    } else {
                        writer.nextFrame(buffer);
                    }
                } catch (Exception e) {
                    if (LOGGER.isLoggable(Level.SEVERE)) {
                        LOGGER.severe("Unable to write frame " + " on behalf of " + nodePushable.getDisplayName()
                                + ":\n" + e);
                    }
                }
                if (frames.size() > 0) {
                    for (ByteBuffer buf : frames) {
                        if (LOGGER.isLoggable(Level.WARNING)) {
                            LOGGER.warning("Flusing OLD frame: " + buf + " on behalf of "
                                    + nodePushable.getDisplayName());
                        }
                        writer.nextFrame(buf);
                    }
                }
{'fraction_non_alphanumeric': 0.08430936017810474, 'fraction_numerical': 0.001161552608653567, 'mean_word_length': 2.071343638525565, 'pattern_counts': {'":': 1, '<': 13, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 3, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5940111', 'n_tokens_mistral': 2896, 'n_tokens_neox': 2743, 'n_words': 570}