<commit_before>/* This file is part of VoltDB.
 * Copyright (C) 2008-2013 VoltDB Inc.
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as
 * published by the Free Software Foundation, either version 3 of the
 * License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with VoltDB.  If not, see <http://www.gnu.org/licenses/>.
 */

package org.voltdb.iv2;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.concurrent.ExecutionException;

import org.apache.zookeeper_voltpatches.KeeperException;
import org.apache.zookeeper_voltpatches.ZooKeeper;
import org.apache.zookeeper_voltpatches.data.Stat;
import org.voltcore.messaging.Mailbox;
import org.voltcore.messaging.TransactionInfoBaseMessage;
import org.voltcore.utils.CoreUtils;
import org.voltcore.utils.Pair;
import org.voltdb.SiteProcedureConnection;
import org.voltdb.SnapshotCompletionInterest.SnapshotCompletionEvent;
import org.voltdb.VoltDB;
import org.voltdb.VoltZK;
import org.voltdb.messaging.FragmentTaskMessage;
import org.voltdb.messaging.Iv2InitiateTaskMessage;
import org.voltdb.messaging.RejoinMessage;
import org.voltdb.rejoin.StreamSnapshotSink;
import org.voltdb.rejoin.TaskLog;

import com.google.common.util.concurrent.SettableFuture;

public class ElasticJoinProducer extends JoinProducerBase implements TaskLog {
    // true if the site has received the first fragment task message
    private boolean m_receivedFirstFragment = false;
    // true if the site has notified the coordinator about the receipt of the first fragment
    // message
    private boolean m_firstFragResponseSent = false;

    // data transfer snapshot completion monitor
    private final SettableFuture<SnapshotCompletionEvent> m_snapshotCompletionMonitor =
            SettableFuture.create();

    // a snapshot sink used to stream table data from multiple sources
    private final StreamSnapshotSink m_dataSink;
    private final Mailbox m_streamSnapshotMb;
    private Long m_partitionTxnId;
    private Stat m_partitionTxnIdStat = new Stat();
    private long m_partitionTxnIds[];

    private class CompletionAction extends JoinCompletionAction {
        @Override
        public void run()
        {
            // TODO: no-op for now
        }
    }

    public ElasticJoinProducer(int partitionId, SiteTaskerQueue taskQueue)
    {
        super(partitionId, "Elastic join producer:" + partitionId + " ", taskQueue);
        m_completionAction = new CompletionAction();
        m_streamSnapshotMb = VoltDB.instance().getHostMessenger().createMailbox();
        m_dataSink = new StreamSnapshotSink(m_streamSnapshotMb);
    }

    /*
     * Inherit the per partition txnid from the long since gone
     * partition that existed in the past
     */
    private void fetchPerPartitionTxnId() {
        ZooKeeper zk = VoltDB.instance().getHostMessenger().getZK();
        byte partitionTxnIdsBytes[] = null;
        try {
            partitionTxnIdsBytes = zk.getData(VoltZK.perPartitionTxnIds, false, m_partitionTxnIdStat);
        } catch (KeeperException.NoNodeException e){return;}//Can be no node if the cluster was never restored
        catch (Exception e) {
            VoltDB.crashLocalVoltDB("Error retrieving per partition txn ids", true, e);
        }
        ByteBuffer buf = ByteBuffer.wrap(partitionTxnIdsBytes);

        int count = buf.getInt();
        m_partitionTxnIds = new long[count];
        for (int ii = 0; ii < count; ii++) {
            long txnId = buf.getLong();
            m_partitionTxnIds[ii] = txnId;
            int partitionId = TxnEgo.getPartitionId(txnId);
            if (partitionId == m_partitionId) {
                m_partitionTxnId = txnId;
                return;
            }
        }
        if (m_partitionTxnId == null) {
            m_partitionTxnIds = null;
            m_partitionTxnIdStat = null;
        }
        System.out.println("Fetched " + count + " txnids and mine is " + m_partitionTxnId);
    }

    /*
     * Set the txnid on the site, clean up the ZK state by racing to remove self
     */
    private void applyAndCleanupPerPartitionTxnId(SiteProcedureConnection connection) {
        //If there was no ID nothing to do
        if (m_partitionTxnId == null) return;
        connection.setPerPartitionTxnIds(m_partitionTxnIds, true);
        ZooKeeper zk = VoltDB.instance().getHostMessenger().getZK();
        do {
            ByteBuffer txnIdsBuf = ByteBuffer.allocate(4 + (8 * (m_partitionTxnIds.length - 1)));
            txnIdsBuf.putInt(m_partitionTxnIds.length - 1);<commit_msg>For ENG-5311 Checkpoint a bug fix, was skipping populating the rest of the ids<commit_after>/* This file is part of VoltDB.
 * Copyright (C) 2008-2013 VoltDB Inc.
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as
 * published by the Free Software Foundation, either version 3 of the
 * License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with VoltDB.  If not, see <http://www.gnu.org/licenses/>.
 */

package org.voltdb.iv2;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.concurrent.ExecutionException;

import org.apache.zookeeper_voltpatches.KeeperException;
import org.apache.zookeeper_voltpatches.ZooKeeper;
import org.apache.zookeeper_voltpatches.data.Stat;
import org.voltcore.messaging.Mailbox;
import org.voltcore.messaging.TransactionInfoBaseMessage;
import org.voltcore.utils.CoreUtils;
import org.voltcore.utils.Pair;
import org.voltdb.SiteProcedureConnection;
import org.voltdb.SnapshotCompletionInterest.SnapshotCompletionEvent;
import org.voltdb.VoltDB;
import org.voltdb.VoltZK;
import org.voltdb.messaging.FragmentTaskMessage;
import org.voltdb.messaging.Iv2InitiateTaskMessage;
import org.voltdb.messaging.RejoinMessage;
import org.voltdb.rejoin.StreamSnapshotSink;
import org.voltdb.rejoin.TaskLog;

import com.google.common.util.concurrent.SettableFuture;

public class ElasticJoinProducer extends JoinProducerBase implements TaskLog {
    // true if the site has received the first fragment task message
    private boolean m_receivedFirstFragment = false;
    // true if the site has notified the coordinator about the receipt of the first fragment
    // message
    private boolean m_firstFragResponseSent = false;

    // data transfer snapshot completion monitor
    private final SettableFuture<SnapshotCompletionEvent> m_snapshotCompletionMonitor =
            SettableFuture.create();

    // a snapshot sink used to stream table data from multiple sources
    private final StreamSnapshotSink m_dataSink;
    private final Mailbox m_streamSnapshotMb;
    private Long m_partitionTxnId;
    private Stat m_partitionTxnIdStat = new Stat();
    private long m_partitionTxnIds[];

    private class CompletionAction extends JoinCompletionAction {
        @Override
        public void run()
        {
            // TODO: no-op for now
        }
    }

    public ElasticJoinProducer(int partitionId, SiteTaskerQueue taskQueue)
    {
        super(partitionId, "Elastic join producer:" + partitionId + " ", taskQueue);
        m_completionAction = new CompletionAction();
        m_streamSnapshotMb = VoltDB.instance().getHostMessenger().createMailbox();
        m_dataSink = new StreamSnapshotSink(m_streamSnapshotMb);
    }

    /*
     * Inherit the per partition txnid from the long since gone
     * partition that existed in the past
     */
    private void fetchPerPartitionTxnId() {
        ZooKeeper zk = VoltDB.instance().getHostMessenger().getZK();
        byte partitionTxnIdsBytes[] = null;
        try {
            partitionTxnIdsBytes = zk.getData(VoltZK.perPartitionTxnIds, false, m_partitionTxnIdStat);
        } catch (KeeperException.NoNodeException e){return;}//Can be no node if the cluster was never restored
        catch (Exception e) {
            VoltDB.crashLocalVoltDB("Error retrieving per partition txn ids", true, e);
        }
        ByteBuffer buf = ByteBuffer.wrap(partitionTxnIdsBytes);

        int count = buf.getInt();
        m_partitionTxnIds = new long[count];
        for (int ii = 0; ii < count; ii++) {
            long txnId = buf.getLong();
            m_partitionTxnIds[ii] = txnId;
            int partitionId = TxnEgo.getPartitionId(txnId);
            System.out.println("Saw partition id " + partitionId + " mine is " + m_partitionId);
            if (partitionId == m_partitionId) {
                m_partitionTxnId = txnId;
                continue;
            }
        }
        if (m_partitionTxnId == null) {
            m_partitionTxnIds = null;
            m_partitionTxnIdStat = null;
        }
        System.out.println("Fetched " + count + " txnids and mine is " + m_partitionTxnId);
    }

    /*
     * Set the txnid on the site, clean up the ZK state by racing to remove self
     */
    private void applyAndCleanupPerPartitionTxnId(SiteProcedureConnection connection) {
        //If there was no ID nothing to do
        if (m_partitionTxnId == null) return;
        connection.setPerPartitionTxnIds(m_partitionTxnIds, true);
        ZooKeeper zk = VoltDB.instance().getHostMessenger().getZK();
        do {
            ByteBuffer txnIdsBuf = ByteBuffer.allocate(4 + (8 * (m_partitionTxnIds.length - 1)));
            txnIdsBuf.putInt(m_partitionTxnIds.length - 1);
{'fraction_non_alphanumeric': 0.0778086542278682, 'fraction_numerical': 0.003572846367606193, 'mean_word_length': 3.2127926421404682, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 7, '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': '2886374', 'n_tokens_mistral': 2923, 'n_tokens_neox': 2846, 'n_words': 843}