<commit_before>/*
 * Copyright (c) 2003-2012 MarkLogic Corporation. All rights reserved.
 */
package com.marklogic.mapreduce;

import java.io.IOException;
import java.net.URI;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import com.marklogic.xcc.ContentSource;
import com.marklogic.xcc.Session;
import com.marklogic.xcc.Session.TransactionMode;
import com.marklogic.xcc.exceptions.RequestException;
import com.marklogic.xcc.exceptions.XccConfigException;

/**
 * A RecordWriter that persists MarkLogicRecord to MarkLogic server.
 * 
 * @author jchen
 *
 */
public abstract class MarkLogicRecordWriter<KEYOUT, VALUEOUT>
extends RecordWriter<KEYOUT, VALUEOUT> implements MarkLogicConstants {
    public static final Log LOG =
        LogFactory.getLog(MarkLogicRecordWriter.class);

    /**
     * Server URI.
     */
    private URI serverUri;
    /**
     * Session to the MarkLogic server.
     */
    private Session session;
    private int count = 0;
    private Configuration conf;
    protected int txnSize;
    
    public MarkLogicRecordWriter(URI serverUri, Configuration conf) {
        this.serverUri = serverUri;
        this.conf = conf;
        this.txnSize = getTransactionSize(conf);
    }
    
    @Override
    public void close(TaskAttemptContext context) throws IOException,
            InterruptedException {
        if (session != null) {
            try {
                if (count > 0 && txnSize > 1) {
                    session.commit();
                }
                session.close();
            } catch (RequestException e) {
                LOG.error(e);
            }
        }
    }
    
    /**
     * Get the session for this writer.  One writer only maintains one session.
     * 
     * @return Session for this writer.
     * @throws IOException
     */
    protected Session getSession() throws IOException {
        if (session == null) {
            // start a session
            try {
                ContentSource cs = InternalUtilities.getOutputContentSource(
                        conf, serverUri);
                session = cs.newSession();
                if (txnSize > 1) {
                    session.setTransactionMode(TransactionMode.UPDATE);
                }
            } catch (XccConfigException e) {
                LOG.error("Error creating a new session: ", e);
                throw new IOException(e);
            }    
        }
        return session;
    }
    
    protected void commitIfNecessary() throws RequestException {
        if (++count == txnSize && txnSize > 1) {
            session.commit();
            count = 0;
        }
    }
       
    public int getTransactionSize(Configuration conf) {
        return 1000;
    }
}<commit_msg>Bug:16209 - Make transactionsize configurable.<commit_after>/*
 * Copyright (c) 2003-2012 MarkLogic Corporation. All rights reserved.
 */
package com.marklogic.mapreduce;

import java.io.IOException;
import java.net.URI;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import com.marklogic.xcc.ContentSource;
import com.marklogic.xcc.Session;
import com.marklogic.xcc.Session.TransactionMode;
import com.marklogic.xcc.exceptions.RequestException;
import com.marklogic.xcc.exceptions.XccConfigException;

/**
 * A RecordWriter that persists MarkLogicRecord to MarkLogic server.
 * 
 * @author jchen
 *
 */
public abstract class MarkLogicRecordWriter<KEYOUT, VALUEOUT>
extends RecordWriter<KEYOUT, VALUEOUT> implements MarkLogicConstants {
    public static final Log LOG =
        LogFactory.getLog(MarkLogicRecordWriter.class);

    /**
     * Server URI.
     */
    private URI serverUri;
    /**
     * Session to the MarkLogic server.
     */
    private Session session;
    private int count = 0;
    private Configuration conf;
    protected int txnSize;
    
    public MarkLogicRecordWriter(URI serverUri, Configuration conf) {
        this.serverUri = serverUri;
        this.conf = conf;
        this.txnSize = getTransactionSize(conf);
    }
    
    @Override
    public void close(TaskAttemptContext context) throws IOException,
            InterruptedException {
        if (session != null) {
            try {
                if (count > 0 && txnSize > 1) {
                    session.commit();
                }
                session.close();
            } catch (RequestException e) {
                LOG.error(e);
            }
        }
    }
    
    /**
     * Get the session for this writer.  One writer only maintains one session.
     * 
     * @return Session for this writer.
     * @throws IOException
     */
    protected Session getSession() throws IOException {
        if (session == null) {
            // start a session
            try {
                ContentSource cs = InternalUtilities.getOutputContentSource(
                        conf, serverUri);
                session = cs.newSession();
                if (txnSize > 1) {
                    session.setTransactionMode(TransactionMode.UPDATE);
                }
            } catch (XccConfigException e) {
                LOG.error("Error creating a new session: ", e);
                throw new IOException(e);
            }    
        }
        return session;
    }
    
    protected void commitIfNecessary() throws RequestException {
        if (++count == txnSize && txnSize > 1) {
            session.commit();
            count = 0;
        }
    }
       
    public int getTransactionSize(Configuration conf) {
        return conf.getInt(TXN_SIZE, 1000);
    }
}
{'fraction_non_alphanumeric': 0.08968686181075562, 'fraction_numerical': 0.006977535738597686, 'mean_word_length': 2.072138003136435, 'pattern_counts': {'":': 0, '<': 7, '<?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': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1567950', 'n_tokens_mistral': 1633, 'n_tokens_neox': 1544, 'n_words': 390}