<commit_before>/**
 * Copyright (C) 2012 Ness Computing, Inc.
 *
 * 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 com.nesscomputing.hbase;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;

import javax.annotation.Nonnull;
import javax.annotation.Nullable;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Function;
import com.google.common.base.Functions;
import com.google.common.base.Preconditions;
import com.google.common.base.Throwables;
import com.google.common.collect.Lists;
import com.google.common.io.Closeables;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.skife.config.TimeSpan;
import org.weakref.jmx.Managed;

import com.nesscomputing.logging.Log;

/**
 * Batch writer to HBase.  Enqueues either as its buffer fills or periodically, to bound the amount of lost information.
 *
 * May only enqueue events while the writer is running.
 */
public class HBaseWriter implements Runnable
{
    private static final Log LOG = Log.findLog();

    private static final Function<Callable<Put>, Put> CALLABLE_FUNCTION = new Function<Callable<Put>, Put>() {
        @Override
        public Put apply(@Nullable final Callable<Put> callable) {
            try {
                return callable == null ? null : callable.call();
            }
            catch (Exception e) {
                throw Throwables.propagate(e);
            }
        }
    };

    private static final Function<Put, byte []> SPILL_FUNCTION = new BinaryConverter.PutToBinary();

    /** Queue of pending writes.  */
    private final LinkedBlockingQueue<Callable<Put>> writeQueue;

    private AtomicBoolean taskRunning = new AtomicBoolean(true);

    /** Holds a reference to the HBase table that this writer uses. */
    private final AtomicReference<HTable> table = new AtomicReference<HTable>(null);

    private final AtomicLong cooloffTime = new AtomicLong(-1L);

    private final AtomicReference<Thread> writerThread = new AtomicReference<Thread>(null);

    private final String writerName;
    private final HBaseWriterConfig hbaseWriterConfig;
    private final Configuration hadoopConfig;

    private final TimeSpan enqueueTimeout;

    private final AtomicLong opsEnqueued = new AtomicLong(0L);
    private final AtomicLong opsEnqSpilled = new AtomicLong(0L);
    private final AtomicLong opsDeqSpilled = new AtomicLong(0L);
    private final AtomicLong opsEnqTimeout = new AtomicLong(0L);
    private final AtomicLong opsEnqCooloff = new AtomicLong(0L);
    private final AtomicLong opsDequeued = new AtomicLong(0L);
    private final AtomicLong opsSent = new AtomicLong(0L);
    private final AtomicLong opsLost = new AtomicLong(0L);
    private final AtomicLong spillsOk = new AtomicLong(0L);
    private final AtomicLong spillsFailed = new AtomicLong(0L);
    private final AtomicInteger longestBurst = new AtomicInteger(0);

    private final File spillingDirectory;
    private final String spillingId;
    private final AtomicInteger spillingCount = new AtomicInteger(0);

    private int backoff = 1;

    private final String tableName;

    HBaseWriter(@Nonnull final String writerName,
                @Nonnull final HBaseWriterConfig hbaseWriterConfig,
                @Nonnull final Configuration hadoopConfig)
    {
        this.writerName = writerName;
        this.hbaseWriterConfig = hbaseWriterConfig;
        this.hadoopConfig = hadoopConfig;

        this.writeQueue = new LinkedBlockingQueue<Callable<Put>>(hbaseWriterConfig.getQueueLength());
        this.enqueueTimeout = hbaseWriterConfig.getEnqueueTimeout();
        this.tableName = hbaseWriterConfig.getTableName();

        final File spillingDirectory = hbaseWriterConfig.getSpillingDirectory();

        if (hbaseWriterConfig.isSpillingEnabled() && spillingDirectory != null) {

            if (!spillingDirectory.exists() && !spillingDirectory.mkdirs()) {
                LOG.error("Could not create directory '%s', spilling will probably not work!", spillingDirectory);
            }

            if (spillingDirectory.exists()
              && spillingDirectory.isDirectory()
              && spillingDirectory.canWrite()
              && spillingDirectory.canExecute()) {
                this.spillingDirectory = spillingDirectory;
                LOG.info("Spilling enabled for %s, directory is '%s'.", writerName, spillingDirectory.getAbsolutePath());
            }
            else {
                this.spillingDirectory = null;
                LOG.info("Spilling disabled for %s, directory '%s' is not usable!", writerName, spillingDirectory);
            }
        }
        else {
            this.spillingDirectory = null;
            LOG.info("Spilling disabled for %s.", writerName);
        }

        this.spillingId = UUID.randomUUID().toString();
    }

    synchronized void start()
    {
        if (hbaseWriterConfig.isEnabled()) {
            try {
            Preconditions.checkState(writerThread.get() == null, "already started, boldly refusing to start twice!");
            Preconditions.checkState(table.get() == null, "Already have a htable object, something went very wrong!");

            LOG.info("Starting HBase Writer for HBase table %s.", tableName);

            connectHTable();

            final Thread thread = new Thread(this, String.format("hbase-%s-writer", tableName));
            writerThread.set(thread);
            thread.start();
            }
            catch (IOException ioe) {
                LOG.warnDebug(ioe, "Could not start HBase writer for %s", tableName);
            }
        }
    }

    synchronized void stop()
    {
        final Thread thread = writerThread.getAndSet(null);
        if (thread != null) {
            LOG.info("Stopping HBase Writer for table %s.", tableName);
            try {
                taskRunning.set(false);
                thread.interrupt();<commit_msg>don't connect to the HBase table on the start method, this can block for a long time and will stop the lifecycle from completing successfully.<commit_after>/**
 * Copyright (C) 2012 Ness Computing, Inc.
 *
 * 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 com.nesscomputing.hbase;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;

import javax.annotation.Nonnull;
import javax.annotation.Nullable;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Function;
import com.google.common.base.Functions;
import com.google.common.base.Preconditions;
import com.google.common.base.Throwables;
import com.google.common.collect.Lists;
import com.google.common.io.Closeables;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.skife.config.TimeSpan;
import org.weakref.jmx.Managed;

import com.nesscomputing.logging.Log;

/**
 * Batch writer to HBase.  Enqueues either as its buffer fills or periodically, to bound the amount of lost information.
 *
 * May only enqueue events while the writer is running.
 */
public class HBaseWriter implements Runnable
{
    private static final Log LOG = Log.findLog();

    private static final Function<Callable<Put>, Put> CALLABLE_FUNCTION = new Function<Callable<Put>, Put>() {
        @Override
        public Put apply(@Nullable final Callable<Put> callable) {
            try {
                return callable == null ? null : callable.call();
            }
            catch (Exception e) {
                throw Throwables.propagate(e);
            }
        }
    };

    private static final Function<Put, byte []> SPILL_FUNCTION = new BinaryConverter.PutToBinary();

    /** Queue of pending writes.  */
    private final LinkedBlockingQueue<Callable<Put>> writeQueue;

    private AtomicBoolean taskRunning = new AtomicBoolean(true);

    /** Holds a reference to the HBase table that this writer uses. */
    private final AtomicReference<HTable> table = new AtomicReference<HTable>(null);

    private final AtomicLong cooloffTime = new AtomicLong(-1L);

    private final AtomicReference<Thread> writerThread = new AtomicReference<Thread>(null);

    private final String writerName;
    private final HBaseWriterConfig hbaseWriterConfig;
    private final Configuration hadoopConfig;

    private final TimeSpan enqueueTimeout;

    private final AtomicLong opsEnqueued = new AtomicLong(0L);
    private final AtomicLong opsEnqSpilled = new AtomicLong(0L);
    private final AtomicLong opsDeqSpilled = new AtomicLong(0L);
    private final AtomicLong opsEnqTimeout = new AtomicLong(0L);
    private final AtomicLong opsEnqCooloff = new AtomicLong(0L);
    private final AtomicLong opsDequeued = new AtomicLong(0L);
    private final AtomicLong opsSent = new AtomicLong(0L);
    private final AtomicLong opsLost = new AtomicLong(0L);
    private final AtomicLong spillsOk = new AtomicLong(0L);
    private final AtomicLong spillsFailed = new AtomicLong(0L);
    private final AtomicInteger longestBurst = new AtomicInteger(0);

    private final File spillingDirectory;
    private final String spillingId;
    private final AtomicInteger spillingCount = new AtomicInteger(0);

    private int backoff = 1;

    private final String tableName;

    HBaseWriter(@Nonnull final String writerName,
                @Nonnull final HBaseWriterConfig hbaseWriterConfig,
                @Nonnull final Configuration hadoopConfig)
    {
        this.writerName = writerName;
        this.hbaseWriterConfig = hbaseWriterConfig;
        this.hadoopConfig = hadoopConfig;

        this.writeQueue = new LinkedBlockingQueue<Callable<Put>>(hbaseWriterConfig.getQueueLength());
        this.enqueueTimeout = hbaseWriterConfig.getEnqueueTimeout();
        this.tableName = hbaseWriterConfig.getTableName();

        final File spillingDirectory = hbaseWriterConfig.getSpillingDirectory();

        if (hbaseWriterConfig.isSpillingEnabled() && spillingDirectory != null) {

            if (!spillingDirectory.exists() && !spillingDirectory.mkdirs()) {
                LOG.error("Could not create directory '%s', spilling will probably not work!", spillingDirectory);
            }

            if (spillingDirectory.exists()
              && spillingDirectory.isDirectory()
              && spillingDirectory.canWrite()
              && spillingDirectory.canExecute()) {
                this.spillingDirectory = spillingDirectory;
                LOG.info("Spilling enabled for %s, directory is '%s'.", writerName, spillingDirectory.getAbsolutePath());
            }
            else {
                this.spillingDirectory = null;
                LOG.info("Spilling disabled for %s, directory '%s' is not usable!", writerName, spillingDirectory);
            }
        }
        else {
            this.spillingDirectory = null;
            LOG.info("Spilling disabled for %s.", writerName);
        }

        this.spillingId = UUID.randomUUID().toString();
    }

    synchronized void start()
    {
        if (hbaseWriterConfig.isEnabled()) {
            Preconditions.checkState(writerThread.get() == null, "already started, boldly refusing to start twice!");
            Preconditions.checkState(table.get() == null, "Already have a htable object, something went very wrong!");

            LOG.info("Starting HBase Writer for HBase table %s.", tableName);

            final Thread thread = new Thread(this, String.format("hbase-%s-writer", tableName));
            writerThread.set(thread);
            thread.start();
        }
    }

    synchronized void stop()
    {
        final Thread thread = writerThread.getAndSet(null);
        if (thread != null) {
            LOG.info("Stopping HBase Writer for table %s.", tableName);
            try {
                taskRunning.set(false);
                thread.interrupt();
{'fraction_non_alphanumeric': 0.09023377758357003, 'fraction_numerical': 0.003204427936785376, 'mean_word_length': 3.146135265700483, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 31, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '2887548', 'n_tokens_mistral': 3703, 'n_tokens_neox': 3531, 'n_words': 1061}