<commit_before>/*
 * Copyright 2010 Outerthought bvba
 *
 * 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 org.lilyproject.rowlog.impl;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.ColumnPrefixFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.lilyproject.rowlock.RowLock;
import org.lilyproject.rowlock.RowLocker;
import org.lilyproject.rowlog.api.*;
import org.lilyproject.util.io.Closer;

/**
 * See {@link RowLog}
 */
public class RowLogImpl implements RowLog, SubscriptionsObserver, RowLogObserver {

    private static final byte PL_BYTE = (byte)1;
    private static final byte ES_BYTE = (byte)2;
    private static final byte[] SEQ_NR = Bytes.toBytes("SEQNR");
    protected RowLogShard shard; // TODO: We only work with one shard for now
    private final HTableInterface rowTable;
    private final byte[] rowLogColumnFamily;
    private RowLogConfig rowLogConfig;
    
    private Map<String, RowLogSubscription> subscriptions = Collections.synchronizedMap(new HashMap<String, RowLogSubscription>());
    protected final String id;
    private RowLogProcessorNotifier processorNotifier = null;
    private Log log = LogFactory.getLog(getClass());
    private RowLogConfigurationManager rowLogConfigurationManager;

    private final AtomicBoolean initialSubscriptionsLoaded = new AtomicBoolean(false);
    private final AtomicBoolean initialRowLogConfigLoaded = new AtomicBoolean(false);
    private final RowLocker rowLocker;
    private byte[] payloadPrefix;
    private byte[] executionStatePrefix;
    private byte[] seqNrQualifier;

    /**
     * The RowLog should be instantiated with information about the table that contains the rows the messages are 
     * related to, and the column families it can use within this table to put the payload and execution state of the
     * messages on.
     * @param rowTable the HBase table containing the rows to which the messages are related
     * @param rowLogColumnFamily the column family in which the payload and execution state of the messages can be stored
     * @param rowLogId a byte uniquely identifying the rowLog amongst all rowLogs in the system
     * @param rowLocker if given, the rowlog will take locks at row level; if null, the locks will be taken at executionstate level
     * @throws RowLogException
     */
    public RowLogImpl(String id, HTableInterface rowTable, byte[] rowLogColumnFamily, byte rowLogId,
            RowLogConfigurationManager rowLogConfigurationManager, RowLocker rowLocker) throws InterruptedException {
        this.id = id;
        this.rowTable = rowTable;
        this.rowLogColumnFamily = rowLogColumnFamily;
        this.payloadPrefix = new byte[]{rowLogId, PL_BYTE};
        this.executionStatePrefix = new byte[]{rowLogId, ES_BYTE};
        this.seqNrQualifier = Bytes.add(new byte[]{rowLogId}, SEQ_NR);
        this.rowLogConfigurationManager = rowLogConfigurationManager;
        this.rowLocker = rowLocker;
        rowLogConfigurationManager.addRowLogObserver(id, this);
        synchronized (initialRowLogConfigLoaded) {
            while(!initialRowLogConfigLoaded.get()) {
                initialRowLogConfigLoaded.wait();
            }
        }
        this.processorNotifier = new RowLogProcessorNotifier(rowLogConfigurationManager, rowLogConfig.getNotifyDelay());
        rowLogConfigurationManager.addSubscriptionsObserver(id, this);
        synchronized (initialSubscriptionsLoaded) {
            while (!initialSubscriptionsLoaded.get()) {
                initialSubscriptionsLoaded.wait();
            }
        }
    }

    public void stop() {
        rowLogConfigurationManager.removeRowLogObserver(id, this);
        synchronized (initialRowLogConfigLoaded) {
            initialRowLogConfigLoaded.set(false);
        }
        rowLogConfigurationManager.removeSubscriptionsObserver(id, this);
        synchronized (initialSubscriptionsLoaded) {
            initialSubscriptionsLoaded.set(false);
        }
        Closer.close(processorNotifier);
    }
    
    @Override
    protected void finalize() throws Throwable {
        stop();
        super.finalize();
    }
    
    public String getId() {
        return id;
    }
    
    public void registerShard(RowLogShard shard) {
        this.shard = shard;
    }
    
    public void unRegisterShard(RowLogShard shard) {
        this.shard = null;
    }
    
    private void putPayload(long seqnr, byte[] payload, long timestamp, Put put) throws IOException {
        put.add(rowLogColumnFamily, payloadQualifier(seqnr, timestamp), payload);
    }

    public byte[] getPayload(RowLogMessage message) throws RowLogException {
        byte[] rowKey = message.getRowKey();
        byte[] qualifier = payloadQualifier(message.getSeqNr(), message.getTimestamp());
        Get get = new Get(rowKey);
        get.addColumn(rowLogColumnFamily, qualifier);
        Result result;
        try {
            result = rowTable.get(get);
        } catch (IOException e) {
            throw new RowLogException("Exception while getting payload from the rowTable", e);
        }
        return result.getValue(rowLogColumnFamily, qualifier);
    }

    public RowLogMessage putMessage(byte[] rowKey, byte[] data, byte[] payload, Put put) throws InterruptedException, RowLogException {
        try {
            // Take current snapshot of the subscriptions so that shard.putMessage and initializeSubscriptions
            // use the exact same set of subscriptions.
            List<RowLogSubscription> subscriptions = getSubscriptions();
            if (subscriptions.isEmpty()) 
                return null;

            // Get a sequence number for this new message
            long seqnr = rowTable.incrementColumnValue(rowKey, rowLogColumnFamily, seqNrQualifier, 1L);

            // Create Put object if not supplied
            boolean ownPut = false;
            if (put == null) {
                put = new Put(rowKey);
                ownPut = true;
            }

            long now = System.currentTimeMillis();

            putPayload(seqnr, payload, now, put);
                    
            RowLogMessage message = new RowLogMessageImpl(now, rowKey, seqnr, data, payload, this);

            putMessageOnShard(message, subscriptions);

            initializeSubscriptions(message, put, subscriptions);

            // If the Put was not supplied by the user, apply it now
            if (ownPut) {
                rowTable.put(put);
            }

            if (rowLogConfig.isEnableNotify()) {
                processorNotifier.notifyProcessor(id, getShard().getId());
            }
            return message;
        } catch (IOException e) {
            throw new RowLogException("Failed to put message on RowLog", e);
        }
    }

    protected void putMessageOnShard(RowLogMessage message, List<RowLogSubscription> subscriptions)
            throws RowLogException {
        List<String> subscriptionIds = new ArrayList<String>(subscriptions.size());
        for (RowLogSubscription subscription : subscriptions) {
            subscriptionIds.add(subscription.getId());
        }
        getShard().putMessage(message, subscriptionIds);
    }

    
    private void initializeSubscriptions(RowLogMessage message, Put put, List<RowLogSubscription> subscriptions)
            throws IOException {
        SubscriptionExecutionState executionState = new SubscriptionExecutionState(message.getTimestamp());
        for (RowLogSubscription subscription : subscriptions) {
            executionState.setState(subscription.getId(), false);
        }
        byte[] qualifier = executionStateQualifier(message.getSeqNr(), message.getTimestamp());
        put.add(rowLogColumnFamily, qualifier, executionState.toBytes());
    }

    public boolean processMessage(RowLogMessage message, RowLock lock) throws RowLogException, InterruptedException {
        if (message == null)
            return true;
        byte[] rowKey = message.getRowKey();
        byte[] executionStateQualifier = executionStateQualifier(message.getSeqNr(), message.getTimestamp());
        Get get = new Get(rowKey);
        get.addColumn(rowLogColumnFamily, executionStateQualifier);
        try {
            Result result = rowTable.get(get);
            if (result.isEmpty()) {
                // No execution state was found indicating an orphan message on the global queue table
                // Treat this message as if it was processed
                return true;
            }
            byte[] previousValue = result.getValue(rowLogColumnFamily, executionStateQualifier);
            SubscriptionExecutionState executionState = SubscriptionExecutionState.fromBytes(previousValue);

            boolean allDone = processMessage(message, executionState);
            
            if (allDone) {
                return handleAllDone(message, rowKey, executionStateQualifier, previousValue, lock);
            } else {
                if (rowLocker != null) {
                    updateExecutionState(rowKey, executionStateQualifier, executionState, previousValue, lock);
                } else {
                    updateExecutionState(rowKey, executionStateQualifier, executionState, previousValue);
                }
                return false;
            }
        } catch (IOException e) {
            throw new RowLogException("Failed to process message", e);
        }
    }

    protected boolean handleAllDone(RowLogMessage message, byte[] rowKey, byte[] executionStateQualifier, byte[] previousValue, RowLock lock) throws RowLogException, IOException {
        if (lock != null) {
            return removeExecutionStateAndPayload(rowKey, executionStateQualifier, payloadQualifier(message.getSeqNr(), message.getTimestamp()), previousValue, lock);
        } else {
            return removeExecutionStateAndPayload(rowKey, executionStateQualifier, payloadQualifier(message.getSeqNr(), message.getTimestamp()), previousValue);
        }
    }

    private boolean processMessage(RowLogMessage message, SubscriptionExecutionState executionState) throws RowLogException, InterruptedException {
        boolean allDone = true;
        List<RowLogSubscription> subscriptionsSnapshot = getSubscriptions();
        if (rowLogConfig.isRespectOrder()) {
            Collections.sort(subscriptionsSnapshot);
        }

        List<RowLogSubscription> subscriptions = getSubscriptions();

        if (log.isDebugEnabled()) {
            log.debug("Processing msg '" + formatId(message) + "' nr of subscriptions: " + subscriptions.size());
        }

        for (RowLogSubscription subscription : subscriptions) {
            String subscriptionId = subscription.getId();

            if (log.isDebugEnabled()) {
                log.debug("Processing msg '" + formatId(message) + "', subscr '" + subscriptionId + "' state: " +
                        executionState.getState(subscriptionId));
            }

            if (!executionState.getState(subscriptionId)) {
                boolean done = false;
                RowLogMessageListener listener = RowLogMessageListenerMapping.INSTANCE.get(subscriptionId);
                if (listener != null) {
                    done = listener.processMessage(message);
                    if (log.isDebugEnabled()) {
                        log.debug("Processing msg '" + formatId(message) + "', subscr '" + subscriptionId +
                                "' processing result: " + done);
                    }
                } else {
                    if (log.isDebugEnabled()) {
                        log.debug("Processing msg '" + formatId(message) + "', subscr '" + subscriptionId +
                                "' no listener present.");
                    }
                }
                executionState.setState(subscriptionId, done);
                if (!done) {
                    allDone = false;
                    if (rowLogConfig.isRespectOrder()) {
                        break;
                    }
                } else {
                    removeMessageFromShard(message, subscriptionId);
                }
            }
        }
        return allDone;
    }

    private String formatId(RowLogMessage message) {
        return Bytes.toStringBinary(message.getRowKey()) + ":" + message.getSeqNr();
    }
    
    public List<RowLogSubscription> getSubscriptions() {
        synchronized (subscriptions) {
            return new ArrayList<RowLogSubscription>(subscriptions.values());
        }
    }
    
    public boolean messageDone(RowLogMessage message, String subscriptionId) throws RowLogException, InterruptedException {
        if (rowLocker != null) { // If the rowLocker exists the lock should be a RowLock
            RowLock rowLock = null;
            try {
                rowLock = rowLocker.lockRow(message.getRowKey());
            } catch (IOException e) {
                log.debug("Exception occured while trying to take lock, retrying", e);
                // retry
            }
            while (rowLock == null) {
                Thread.sleep(10);
                try {
                    rowLock = rowLocker.lockRow(message.getRowKey());
                } catch (IOException e) {
                    log.debug("Exception occured while trying to take lock, retrying", e);
                    // retry
                }
            }
            return messageDoneRowLocked(message, subscriptionId, rowLock);
        }
        return messageDone(message, subscriptionId, 0); 
    }
    
    private boolean messageDoneRowLocked(RowLogMessage message, String subscriptionId, RowLock rowLock) throws RowLogException {
        byte[] rowKey = message.getRowKey();
        byte[] executionStateQualifier = executionStateQualifier(message.getSeqNr(), message.getTimestamp());
        Get get = new Get(rowKey);
        get.addColumn(rowLogColumnFamily, executionStateQualifier);
        try {
            Result result = rowTable.get(get);
            if (!result.isEmpty()) {
                byte[] previousValue = result.getValue(rowLogColumnFamily, executionStateQualifier);
                SubscriptionExecutionState executionState = SubscriptionExecutionState.fromBytes(previousValue);
                executionState.setState(subscriptionId, true);
                if (executionState.allDone()) {
                    handleAllDone(message, rowKey, executionStateQualifier, previousValue, rowLock);
                } else {
                    if (!updateExecutionState(rowKey, executionStateQualifier, executionState, previousValue, rowLock))
                        return false;
                }
            }
            removeMessageFromShard(message, subscriptionId);
            return true;
        } catch (IOException e) {
            throw new RowLogException("Failed to put message to done", e);
        }
    
    }
    
    private boolean messageDone(RowLogMessage message, String subscriptionId, int count) throws RowLogException {
        if (count >= 10) {
            return false;
        }
        byte[] rowKey = message.getRowKey();
        byte[] executionStateQualifier = executionStateQualifier(message.getSeqNr(), message.getTimestamp());
        Get get = new Get(rowKey);
        get.addColumn(rowLogColumnFamily, executionStateQualifier);
        try {
            Result result = rowTable.get(get);
            if (!result.isEmpty()) {
                byte[] previousValue = result.getValue(rowLogColumnFamily, executionStateQualifier);
                SubscriptionExecutionState executionState = SubscriptionExecutionState.fromBytes(previousValue);
                executionState.setState(subscriptionId, true);
                if (executionState.allDone()) {
                    handleAllDone(message, rowKey, executionStateQualifier, previousValue, null);
                } else {
                    if (!updateExecutionState(rowKey, executionStateQualifier, executionState, previousValue)) {
                        return messageDone(message, subscriptionId, count+1); // Retry
                    }
                }
            }
            removeMessageFromShard(message, subscriptionId);
            return true;<commit_msg>Adding some TODO's on things i noticed.<commit_after>/*
 * Copyright 2010 Outerthought bvba
 *
 * 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 org.lilyproject.rowlog.impl;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.ColumnPrefixFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.lilyproject.rowlock.RowLock;
import org.lilyproject.rowlock.RowLocker;
import org.lilyproject.rowlog.api.*;
import org.lilyproject.util.io.Closer;

/**
 * See {@link RowLog}
 */
public class RowLogImpl implements RowLog, SubscriptionsObserver, RowLogObserver {

    private static final byte PL_BYTE = (byte)1;
    private static final byte ES_BYTE = (byte)2;
    private static final byte[] SEQ_NR = Bytes.toBytes("SEQNR");
    protected RowLogShard shard; // TODO: We only work with one shard for now
    private final HTableInterface rowTable;
    private final byte[] rowLogColumnFamily;
    private RowLogConfig rowLogConfig;
    
    private Map<String, RowLogSubscription> subscriptions = Collections.synchronizedMap(new HashMap<String, RowLogSubscription>());
    protected final String id;
    private RowLogProcessorNotifier processorNotifier = null;
    private Log log = LogFactory.getLog(getClass());
    private RowLogConfigurationManager rowLogConfigurationManager;

    private final AtomicBoolean initialSubscriptionsLoaded = new AtomicBoolean(false);
    private final AtomicBoolean initialRowLogConfigLoaded = new AtomicBoolean(false);
    private final RowLocker rowLocker;
    private byte[] payloadPrefix;
    private byte[] executionStatePrefix;
    private byte[] seqNrQualifier;

    /**
     * The RowLog should be instantiated with information about the table that contains the rows the messages are 
     * related to, and the column families it can use within this table to put the payload and execution state of the
     * messages on.
     * @param rowTable the HBase table containing the rows to which the messages are related
     * @param rowLogColumnFamily the column family in which the payload and execution state of the messages can be stored
     * @param rowLogId a byte uniquely identifying the rowLog amongst all rowLogs in the system
     * @param rowLocker if given, the rowlog will take locks at row level; if null, the locks will be taken at executionstate level
     * @throws RowLogException
     */
    public RowLogImpl(String id, HTableInterface rowTable, byte[] rowLogColumnFamily, byte rowLogId,
            RowLogConfigurationManager rowLogConfigurationManager, RowLocker rowLocker) throws InterruptedException {
        this.id = id;
        this.rowTable = rowTable;
        this.rowLogColumnFamily = rowLogColumnFamily;
        this.payloadPrefix = new byte[]{rowLogId, PL_BYTE};
        this.executionStatePrefix = new byte[]{rowLogId, ES_BYTE};
        this.seqNrQualifier = Bytes.add(new byte[]{rowLogId}, SEQ_NR);
        this.rowLogConfigurationManager = rowLogConfigurationManager;
        this.rowLocker = rowLocker;
        rowLogConfigurationManager.addRowLogObserver(id, this);
        synchronized (initialRowLogConfigLoaded) {
            while(!initialRowLogConfigLoaded.get()) {
                initialRowLogConfigLoaded.wait();
            }
        }
        this.processorNotifier = new RowLogProcessorNotifier(rowLogConfigurationManager, rowLogConfig.getNotifyDelay());
        rowLogConfigurationManager.addSubscriptionsObserver(id, this);
        synchronized (initialSubscriptionsLoaded) {
            while (!initialSubscriptionsLoaded.get()) {
                initialSubscriptionsLoaded.wait();
            }
        }
    }

    public void stop() {
        rowLogConfigurationManager.removeRowLogObserver(id, this);
        synchronized (initialRowLogConfigLoaded) {
            initialRowLogConfigLoaded.set(false);
        }
        rowLogConfigurationManager.removeSubscriptionsObserver(id, this);
        synchronized (initialSubscriptionsLoaded) {
            initialSubscriptionsLoaded.set(false);
        }
        Closer.close(processorNotifier);
    }
    
    @Override
    protected void finalize() throws Throwable {
        stop();
        super.finalize();
    }
    
    public String getId() {
        return id;
    }
    
    public void registerShard(RowLogShard shard) {
        this.shard = shard;
    }
    
    public void unRegisterShard(RowLogShard shard) {
        this.shard = null;
    }
    
    private void putPayload(long seqnr, byte[] payload, long timestamp, Put put) throws IOException {
        put.add(rowLogColumnFamily, payloadQualifier(seqnr, timestamp), payload);
    }

    public byte[] getPayload(RowLogMessage message) throws RowLogException {
        byte[] rowKey = message.getRowKey();
        byte[] qualifier = payloadQualifier(message.getSeqNr(), message.getTimestamp());
        Get get = new Get(rowKey);
        get.addColumn(rowLogColumnFamily, qualifier);
        Result result;
        try {
            result = rowTable.get(get);
        } catch (IOException e) {
            throw new RowLogException("Exception while getting payload from the rowTable", e);
        }
        return result.getValue(rowLogColumnFamily, qualifier);
    }

    public RowLogMessage putMessage(byte[] rowKey, byte[] data, byte[] payload, Put put) throws InterruptedException, RowLogException {
        try {
            // Take current snapshot of the subscriptions so that shard.putMessage and initializeSubscriptions
            // use the exact same set of subscriptions.
            List<RowLogSubscription> subscriptions = getSubscriptions();
            if (subscriptions.isEmpty()) 
                return null;

            // Get a sequence number for this new message
            long seqnr = rowTable.incrementColumnValue(rowKey, rowLogColumnFamily, seqNrQualifier, 1L);

            // Create Put object if not supplied
            boolean ownPut = false;
            if (put == null) {
                put = new Put(rowKey);
                ownPut = true;
            }

            long now = System.currentTimeMillis();

            putPayload(seqnr, payload, now, put);
                    
            RowLogMessage message = new RowLogMessageImpl(now, rowKey, seqnr, data, payload, this);

            putMessageOnShard(message, subscriptions);

            initializeSubscriptions(message, put, subscriptions);

            // If the Put was not supplied by the user, apply it now
            if (ownPut) {
                rowTable.put(put);
            }

            if (rowLogConfig.isEnableNotify()) {
                processorNotifier.notifyProcessor(id, getShard().getId());
            }
            return message;
        } catch (IOException e) {
            throw new RowLogException("Failed to put message on RowLog", e);
        }
    }

    protected void putMessageOnShard(RowLogMessage message, List<RowLogSubscription> subscriptions)
            throws RowLogException {
        List<String> subscriptionIds = new ArrayList<String>(subscriptions.size());
        for (RowLogSubscription subscription : subscriptions) {
            subscriptionIds.add(subscription.getId());
        }
        getShard().putMessage(message, subscriptionIds);
    }

    
    private void initializeSubscriptions(RowLogMessage message, Put put, List<RowLogSubscription> subscriptions)
            throws IOException {
        SubscriptionExecutionState executionState = new SubscriptionExecutionState(message.getTimestamp());
        for (RowLogSubscription subscription : subscriptions) {
            executionState.setState(subscription.getId(), false);
        }
        byte[] qualifier = executionStateQualifier(message.getSeqNr(), message.getTimestamp());
        put.add(rowLogColumnFamily, qualifier, executionState.toBytes());
    }

    public boolean processMessage(RowLogMessage message, RowLock lock) throws RowLogException, InterruptedException {
        if (message == null)
            return true;
        byte[] rowKey = message.getRowKey();
        byte[] executionStateQualifier = executionStateQualifier(message.getSeqNr(), message.getTimestamp());
        Get get = new Get(rowKey);
        get.addColumn(rowLogColumnFamily, executionStateQualifier);
        try {
            Result result = rowTable.get(get);
            if (result.isEmpty()) {
                // No execution state was found indicating an orphan message on the global queue table
                // Treat this message as if it was processed
                return true;
            }
            byte[] previousValue = result.getValue(rowLogColumnFamily, executionStateQualifier);
            SubscriptionExecutionState executionState = SubscriptionExecutionState.fromBytes(previousValue);

            boolean allDone = processMessage(message, executionState);
            
            if (allDone) {
                return handleAllDone(message, rowKey, executionStateQualifier, previousValue, lock);
            } else {
                if (rowLocker != null) {
                    // TODO (bruno) return value is ignored, is this ok?
                    updateExecutionState(rowKey, executionStateQualifier, executionState, previousValue, lock);
                } else {
                    // TODO (bruno) return value is ignored, is this ok?
                    updateExecutionState(rowKey, executionStateQualifier, executionState, previousValue);
                }
                return false;
            }
        } catch (IOException e) {
            throw new RowLogException("Failed to process message", e);
        }
    }

    protected boolean handleAllDone(RowLogMessage message, byte[] rowKey, byte[] executionStateQualifier, byte[] previousValue, RowLock lock) throws RowLogException, IOException {
        if (lock != null) {
            return removeExecutionStateAndPayload(rowKey, executionStateQualifier, payloadQualifier(message.getSeqNr(), message.getTimestamp()), previousValue, lock);
        } else {
            return removeExecutionStateAndPayload(rowKey, executionStateQualifier, payloadQualifier(message.getSeqNr(), message.getTimestamp()), previousValue);
        }
    }

    private boolean processMessage(RowLogMessage message, SubscriptionExecutionState executionState) throws RowLogException, InterruptedException {
        boolean allDone = true;
        List<RowLogSubscription> subscriptionsSnapshot = getSubscriptions();
        if (rowLogConfig.isRespectOrder()) {
            Collections.sort(subscriptionsSnapshot);
        }

        List<RowLogSubscription> subscriptions = getSubscriptions();

        if (log.isDebugEnabled()) {
            log.debug("Processing msg '" + formatId(message) + "' nr of subscriptions: " + subscriptions.size());
        }

        for (RowLogSubscription subscription : subscriptions) {
            String subscriptionId = subscription.getId();

            if (log.isDebugEnabled()) {
                log.debug("Processing msg '" + formatId(message) + "', subscr '" + subscriptionId + "' state: " +
                        executionState.getState(subscriptionId));
            }

            if (!executionState.getState(subscriptionId)) {
                boolean done = false;
                RowLogMessageListener listener = RowLogMessageListenerMapping.INSTANCE.get(subscriptionId);
                if (listener != null) {
                    done = listener.processMessage(message);
                    if (log.isDebugEnabled()) {
                        log.debug("Processing msg '" + formatId(message) + "', subscr '" + subscriptionId +
                                "' processing result: " + done);
                    }
                } else {
                    if (log.isDebugEnabled()) {
                        log.debug("Processing msg '" + formatId(message) + "', subscr '" + subscriptionId +
                                "' no listener present.");
                    }
                }
                executionState.setState(subscriptionId, done);
                if (!done) {
                    allDone = false;
                    if (rowLogConfig.isRespectOrder()) {
                        break;
                    }
                } else {
                    removeMessageFromShard(message, subscriptionId);
                }
            }
        }
        return allDone;
    }

    private String formatId(RowLogMessage message) {
        return Bytes.toStringBinary(message.getRowKey()) + ":" + message.getSeqNr();
    }
    
    public List<RowLogSubscription> getSubscriptions() {
        synchronized (subscriptions) {
            return new ArrayList<RowLogSubscription>(subscriptions.values());
        }
    }
    
    public boolean messageDone(RowLogMessage message, String subscriptionId) throws RowLogException, InterruptedException {
        if (rowLocker != null) { // If the rowLocker exists the lock should be a RowLock
            RowLock rowLock = null;
            try {
                rowLock = rowLocker.lockRow(message.getRowKey());
            } catch (IOException e) {
                log.debug("Exception occurred while trying to take lock, retrying", e);
                // retry
            }
            while (rowLock == null) {
                Thread.sleep(10);
                try {
                    rowLock = rowLocker.lockRow(message.getRowKey());
                } catch (IOException e) {
                    log.debug("Exception occurred while trying to take lock, retrying", e);
                    // retry
                }
            }
            return messageDoneRowLocked(message, subscriptionId, rowLock);
        }
        return messageDone(message, subscriptionId, 0); 
    }
    
    private boolean messageDoneRowLocked(RowLogMessage message, String subscriptionId, RowLock rowLock) throws RowLogException {
        byte[] rowKey = message.getRowKey();
        byte[] executionStateQualifier = executionStateQualifier(message.getSeqNr(), message.getTimestamp());
        Get get = new Get(rowKey);
        get.addColumn(rowLogColumnFamily, executionStateQualifier);
        try {
            Result result = rowTable.get(get);
            if (!result.isEmpty()) {
                byte[] previousValue = result.getValue(rowLogColumnFamily, executionStateQualifier);
                SubscriptionExecutionState executionState = SubscriptionExecutionState.fromBytes(previousValue);
                executionState.setState(subscriptionId, true);
                if (executionState.allDone()) {
                    handleAllDone(message, rowKey, executionStateQualifier, previousValue, rowLock);
                } else {
                    if (!updateExecutionState(rowKey, executionStateQualifier, executionState, previousValue, rowLock))
                        return false;
                }
            }
            removeMessageFromShard(message, subscriptionId);
            return true;
        } catch (IOException e) {
            throw new RowLogException("Failed to put message to done", e);
        }
    
    }
    
    private boolean messageDone(RowLogMessage message, String subscriptionId, int count) throws RowLogException {
        if (count >= 10) {
            return false;
        }
        byte[] rowKey = message.getRowKey();
        byte[] executionStateQualifier = executionStateQualifier(message.getSeqNr(), message.getTimestamp());
        Get get = new Get(rowKey);
        get.addColumn(rowLogColumnFamily, executionStateQualifier);
        try {
            Result result = rowTable.get(get);
            if (!result.isEmpty()) {
                byte[] previousValue = result.getValue(rowLogColumnFamily, executionStateQualifier);
                SubscriptionExecutionState executionState = SubscriptionExecutionState.fromBytes(previousValue);
                executionState.setState(subscriptionId, true);
                if (executionState.allDone()) {
                    // TODO (bruno) return value is ignored here, is this fine? Probably yes since everyone else
                    // is done with this row (unless more than one can come to this conclusion at the same time)
                    handleAllDone(message, rowKey, executionStateQualifier, previousValue, null);
                } else {
                    if (!updateExecutionState(rowKey, executionStateQualifier, executionState, previousValue)) {
                        return messageDone(message, subscriptionId, count+1); // Retry
                    }
                }
            }
            removeMessageFromShard(message, subscriptionId);
            return true;
{'fraction_non_alphanumeric': 0.08291091301054594, 'fraction_numerical': 0.0009905028258462973, 'mean_word_length': 2.431670498850345, 'pattern_counts': {'":': 2, '<': 25, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4266774', 'n_tokens_mistral': 8327, 'n_tokens_neox': 8158, 'n_words': 2230}