/**
 * Copyright 2015 Palantir Technologies
 *
 * Licensed under the BSD-3 License (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://opensource.org/licenses/BSD-3-Clause
 *
 * 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 com.palantir.atlasdb.keyvalue.cassandra;

import org.apache.commons.lang.RandomStringUtils;
import org.junit.Ignore;
import org.junit.Test;

import com.palantir.atlasdb.cassandra.CassandraKeyValueServiceConfigManager;
import com.palantir.atlasdb.cleaner.AbstractSweeperTest;
import com.palantir.atlasdb.keyvalue.api.KeyValueService;
import com.palantir.atlasdb.protos.generated.TableMetadataPersistence;

public class CassandraKeyValueServiceSweeperTest extends AbstractSweeperTest {
    @Override
    protected KeyValueService getKeyValueService() {
        return CassandraKeyValueService.create(
                CassandraKeyValueServiceConfigManager.createSimpleManager(CassandraTestSuite.CASSANDRA_KVS_CONFIG), CassandraTestSuite.LEADER_CONFIG);
    }

    @Test
    public void should_not_oom_when_there_are_many_large_values_to_sweep() {
        createTable(TableMetadataPersistence.SweepStrategy.CONSERVATIVE);

        int numInsertions = 100;
        insertMultipleValues(numInsertions);

        completeSweep(numInsertions);
    }

    private void insertMultipleValues(int numInsertions) {
        for (int ts = 0; ts < numInsertions; ts++) {
            System.out.println("putting with ts = " + ts);
            put("row1", makeLongRandomString(), ts);
        }
    }

    private String makeLongRandomString() {
        return RandomStringUtils.random(1_000_000);
    }
}
@@ -632,16 +632,119 @@
 andra;%0A%0A
+import static org.hamcrest.MatcherAssert.assertThat;%0Aimport static org.hamcrest.core.IsEqual.equalTo;%0A%0A
 import o
@@ -789,33 +789,8 @@
 ls;%0A
-import org.junit.Ignore;%0A
 impo
@@ -1001,16 +1001,71 @@
 ervice;%0A
+import com.palantir.atlasdb.keyvalue.api.SweepResults;%0A
 import c
@@ -1650,19 +1650,20 @@
         
-int
+long
  numInse
@@ -1734,21 +1734,175 @@
     
-completeSweep
+long sweepTimestamp = numInsertions + 1;%0A        SweepResults sweepResults = completeSweep(sweepTimestamp);%0A%0A        assertThat(sweepResults.getCellsDeleted(), equalTo
 (num
@@ -1911,16 +1911,21 @@
 sertions
+ - 1)
 );%0A    %7D
@@ -1964,19 +1964,20 @@
 eValues(
-int
+long
  numInse
@@ -2012,15 +2012,16 @@
 s = 
-0
+1
 ; ts %3C
+=
  num

Check that all the cells are actually swept