Solution for OSS Cassandra Bootcamp

Xyz.java

package org.apache.cassandra.db.compaction;

import java.io.IOException;
import java.util.Arrays;
import java.util.Collection;
import java.util.Map;
import java.util.Set;
import java.util.HashSet;
import java.util.List;
import java.util.ArrayList;

import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.sstable.metadata.MetadataType;
import org.apache.cassandra.io.sstable.metadata.CompactionMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class Xyz extends AbstractCompactionStrategy {
    private static final Logger logger = LoggerFactory.getLogger(Xyz.class);

    public Xyz(ColumnFamilyStore cfs, Map<String, String> options) {
        super(cfs, options);
        // TODO Auto-generated constructor stub
        logger.info("In Xyz.<init>");
    }

    @Override
    public AbstractCompactionTask getNextBackgroundTask(int gcBefore) {
        // TODO Auto-generated method stub
        logger.info("In Xyz.getNextBackgroundTask");
        double score = 10000000.0;
        SSTableReader best1 = null;
        SSTableReader best2 = null;
        Set<SSTableReader> sstSet = cfs.getUncompactingSSTables();
        
        for (SSTableReader sst1 : sstSet) {
            for (SSTableReader sst2 : sstSet) {
                if (sst1.getFilename().compareTo(sst2.getFilename()) >= 0)
                    continue;
                Set<SSTableReader> pair = new HashSet<SSTableReader>();
                pair.add(sst1);
                pair.add(sst2);
                double newscore = SSTableReader.estimateCompactionGain(pair);
                if (newscore < score) {
                        best1 = sst1;
                        best2 = sst2;
                        score = newscore;
                }
            }
        }
        if (null != best1) {
            List<SSTableReader> ret = new ArrayList<SSTableReader>();
            ret.add(best1);
            ret.add(best2);
            if (cfs.getDataTracker().markCompacting(ret)) {
                logger.info("Xyz.getNextBackgroundTask - CompactionTask of " + best1.getFilename() + " and " + best2.getFilename());
                return new CompactionTask(cfs, ret, gcBefore, false);
            }
        }

        return null;
    }

    @Override
    public Collection<AbstractCompactionTask> getMaximalTask(int gcBefore) {
        // TODO Auto-generated method stub
        logger.info("In Xyz.getMaximalTask");
        List<SSTableReader> ret = new ArrayList<SSTableReader>();
        Set<SSTableReader> sstSet = cfs.getUncompactingSSTables();
        if (sstSet.isEmpty())
            return null;
        for (SSTableReader sst : sstSet) {
            ret.add(sst);
        }
        if (cfs.getDataTracker().markCompacting(sstSet))
            return Arrays.<AbstractCompactionTask>asList(new CompactionTask(cfs, sstSet, gcBefore, false));
        return null;
    }

    @Override
    public AbstractCompactionTask getUserDefinedTask(
            Collection<SSTableReader> sstables, int gcBefore) {
        // TODO Auto-generated method stub
        return null;
    }

    @Override
    public int getEstimatedRemainingTasks() {
        // TODO Auto-generated method stub
        return 0;
    }

    @Override
    public long getMaxSSTableBytes() {
        // TODO Auto-generated method stub
        return Long.MAX_VALUE;
    }

    @Override
    public void addSSTable(SSTableReader added) {
        // TODO Auto-generated method stub
    }

    @Override
    public void removeSSTable(SSTableReader sstable) {
        // TODO Auto-generated method stub
    }

}