<commit_before>package com.jivesoftware.os.amza.service.replication;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.jivesoftware.os.amza.api.partition.VersionedPartitionName;
import com.jivesoftware.os.amza.service.AmzaRingStoreReader;
import com.jivesoftware.os.amza.service.storage.PartitionCreator;
import com.jivesoftware.os.amza.service.storage.PartitionIndex;
import com.jivesoftware.os.amza.service.storage.PartitionStore;
import com.jivesoftware.os.amza.shared.stats.AmzaStats;
import com.jivesoftware.os.amza.shared.stats.AmzaStats.CompactionFamily;
import com.jivesoftware.os.amza.shared.take.HighwaterStorage;
import com.jivesoftware.os.aquarium.State;
import com.jivesoftware.os.mlogger.core.MetricLogger;
import com.jivesoftware.os.mlogger.core.MetricLoggerFactory;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * @author jonathan.colt
 */
public class PartitionComposter {

    private static final MetricLogger LOG = MetricLoggerFactory.getLogger();

    private ScheduledExecutorService scheduledThreadPool;

    private final AmzaStats amzaStats;
    private final PartitionIndex partitionIndex;
    private final PartitionCreator partitionProvider;
    private final AmzaRingStoreReader amzaRingReader;
    private final PartitionStripeProvider partitionStripeProvider;
    private final PartitionStateStorage partitionStateStorage;

    public PartitionComposter(AmzaStats amzaStats,
        PartitionIndex partitionIndex,
        PartitionCreator partitionProvider,
        AmzaRingStoreReader amzaRingReader,
        PartitionStateStorage partitionStateStorage,
        PartitionStripeProvider partitionStripeProvider) {

        this.amzaStats = amzaStats;
        this.partitionIndex = partitionIndex;
        this.partitionProvider = partitionProvider;
        this.amzaRingReader = amzaRingReader;
        this.partitionStateStorage = partitionStateStorage;
        this.partitionStripeProvider = partitionStripeProvider;
    }

    public void start() throws Exception {

        scheduledThreadPool = Executors.newScheduledThreadPool(1,
            new ThreadFactoryBuilder().setNameFormat("partition-composter-%d").build());
        scheduledThreadPool.scheduleWithFixedDelay(() -> {
            try {
                compost();
            } catch (Exception x) {
                LOG.debug("Failing to compact tombstones.", x);

            }
        }, 0, 1, TimeUnit.MINUTES); // TODO config
    }

    public void stop() throws Exception {
        this.scheduledThreadPool.shutdownNow();
        this.scheduledThreadPool = null;
    }

    public void compost() throws Exception {
        List<VersionedPartitionName> composted = new ArrayList<>();
        partitionStateStorage.streamLocalState((partitionName, ringMember, versionedState) -> {
            if (versionedState.isCurrentState(State.expunged)) {
                try {
                    amzaStats.beginCompaction(CompactionFamily.expunge, partitionName + " " + versionedState);
                    partitionStripeProvider.txPartition(partitionName, (PartitionStripe stripe, HighwaterStorage highwaterStorage) -> {
                        VersionedPartitionName versionedPartitionName = new VersionedPartitionName(partitionName,
                            versionedState.getPartitionVersion());
                        if (stripe.expungePartition(versionedPartitionName)) {
                            PartitionStore partitionStore = partitionIndex.remove(versionedPartitionName);
                            partitionStore.getWalStorage().expunge();
                            highwaterStorage.expunge(versionedPartitionName);
                            composted.add(versionedPartitionName);
                        }
                        return null;
                    });
                } finally {
                    amzaStats.endCompaction(CompactionFamily.expunge, partitionName + " " + versionedState);
                }

            } else if (!amzaRingReader.isMemberOfRing(partitionName.getRingName()) || !partitionProvider.hasPartition(partitionName)) {
                partitionStateStorage.markForDisposal(new VersionedPartitionName(partitionName, versionedState.getPartitionVersion()), ringMember);
            }
            return true;
        });
        partitionStateStorage.expunged(composted);

    }

}<commit_msg>Added loggging around expunging.<commit_after>package com.jivesoftware.os.amza.service.replication;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.jivesoftware.os.amza.api.partition.VersionedPartitionName;
import com.jivesoftware.os.amza.service.AmzaRingStoreReader;
import com.jivesoftware.os.amza.service.storage.PartitionCreator;
import com.jivesoftware.os.amza.service.storage.PartitionIndex;
import com.jivesoftware.os.amza.service.storage.PartitionStore;
import com.jivesoftware.os.amza.shared.stats.AmzaStats;
import com.jivesoftware.os.amza.shared.stats.AmzaStats.CompactionFamily;
import com.jivesoftware.os.amza.shared.take.HighwaterStorage;
import com.jivesoftware.os.aquarium.State;
import com.jivesoftware.os.mlogger.core.MetricLogger;
import com.jivesoftware.os.mlogger.core.MetricLoggerFactory;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * @author jonathan.colt
 */
public class PartitionComposter {

    private static final MetricLogger LOG = MetricLoggerFactory.getLogger();

    private ScheduledExecutorService scheduledThreadPool;

    private final AmzaStats amzaStats;
    private final PartitionIndex partitionIndex;
    private final PartitionCreator partitionProvider;
    private final AmzaRingStoreReader amzaRingReader;
    private final PartitionStripeProvider partitionStripeProvider;
    private final PartitionStateStorage partitionStateStorage;

    public PartitionComposter(AmzaStats amzaStats,
        PartitionIndex partitionIndex,
        PartitionCreator partitionProvider,
        AmzaRingStoreReader amzaRingReader,
        PartitionStateStorage partitionStateStorage,
        PartitionStripeProvider partitionStripeProvider) {

        this.amzaStats = amzaStats;
        this.partitionIndex = partitionIndex;
        this.partitionProvider = partitionProvider;
        this.amzaRingReader = amzaRingReader;
        this.partitionStateStorage = partitionStateStorage;
        this.partitionStripeProvider = partitionStripeProvider;
    }

    public void start() throws Exception {

        scheduledThreadPool = Executors.newScheduledThreadPool(1,
            new ThreadFactoryBuilder().setNameFormat("partition-composter-%d").build());
        scheduledThreadPool.scheduleWithFixedDelay(() -> {
            try {
                compost();
            } catch (Exception x) {
                LOG.debug("Failing to compact tombstones.", x);

            }
        }, 0, 1, TimeUnit.MINUTES); // TODO config
    }

    public void stop() throws Exception {
        this.scheduledThreadPool.shutdownNow();
        this.scheduledThreadPool = null;
    }

    public void compost() throws Exception {
        List<VersionedPartitionName> composted = new ArrayList<>();
        partitionStateStorage.streamLocalState((partitionName, ringMember, versionedState) -> {
            long partitionVersion = versionedState.getPartitionVersion();
            VersionedPartitionName versionedPartitionName = new VersionedPartitionName(partitionName, partitionVersion);
            if (versionedState.isCurrentState(State.expunged)) {
                try {
                    LOG.info("Expunging {} {}.", partitionName, partitionVersion);
                    amzaStats.beginCompaction(CompactionFamily.expunge, partitionName + " " + versionedState);
                    partitionStripeProvider.txPartition(partitionName, (PartitionStripe stripe, HighwaterStorage highwaterStorage) -> {
                        if (stripe.expungePartition(versionedPartitionName)) {
                            PartitionStore partitionStore = partitionIndex.remove(versionedPartitionName);
                            partitionStore.getWalStorage().expunge();
                            highwaterStorage.expunge(versionedPartitionName);
                            composted.add(versionedPartitionName);
                        }
                        return null;
                    });
                    LOG.info("Expunged {} {}.", partitionName, partitionVersion);
                } finally {
                    amzaStats.endCompaction(CompactionFamily.expunge, partitionName + " " + versionedState);
                }

            } else if (!amzaRingReader.isMemberOfRing(partitionName.getRingName())) {
                LOG.info("Marked {} {} for disposal because it not a member of ring {}.", partitionName, partitionVersion, partitionName.getRingName());
                partitionStateStorage.markForDisposal(versionedPartitionName, ringMember);
            } else if (!partitionProvider.hasPartition(partitionName)) {
                LOG.info("Marked {} {} for disposal because  no partition is defined on this node.", partitionName, partitionVersion);
                partitionStateStorage.markForDisposal(versionedPartitionName, ringMember);
            }
            return true;
        });
        partitionStateStorage.expunged(composted);
    }
}
{'fraction_non_alphanumeric': 0.08173931221908644, 'fraction_numerical': 0.0006271558482282847, 'mean_word_length': 3.0184796304073918, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 13, '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': '1378799', 'n_tokens_mistral': 2480, 'n_tokens_neox': 2377, 'n_words': 410}