stack/core/src/test/java/org/apache/usergrid/persistence/CollectionDeleteTest.java - usergrid - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You 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.apache.usergrid.persistence;


 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;

 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.UUID;

 import org.apache.commons.lang.RandomStringUtils;
 import org.apache.usergrid.AbstractCoreIT;
 import org.apache.usergrid.corepersistence.index.CollectionDeleteRequestBuilder;
 import org.apache.usergrid.corepersistence.index.CollectionDeleteService;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;


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


     private static final int ENTITIES_TO_DELETE = 1100;
     private static final int ENTITIES_TO_ADD_AFTER_TIME = 5;


     @Before
     public void startReporting() {

         if (logger.isDebugEnabled()) {
             logger.debug("Starting metrics reporting");
         }
     }


     @After
     public void printReport() {
         logger.debug( "Printing metrics report" );
     }


     @Test( timeout = 240000 )
     public void clearOneCollection() throws Exception {

         logger.info( "Started clearOneCollection()" );

         String rand = RandomStringUtils.randomAlphanumeric( 5 );
         final UUID appId = setup.createApplication( "org_" + rand, "app_" + rand );

         final EntityManager em = setup.getEmf().getEntityManager( appId );

         final CollectionDeleteService collectionDeleteService = setup.getInjector().getInstance( CollectionDeleteService.class );

         // ----------------- create a bunch of entities

         Map<String, Object> entityMap = new HashMap<String, Object>() {{
             put( "key1", 1000 );
             put( "key2", 2000 );
             put( "key3", "Some value" );
         }};

         String collectionName = "items";
         String itemType = "item";


         List<EntityRef> entityRefs = new ArrayList<EntityRef>();
         for ( int i = 0; i < ENTITIES_TO_DELETE; i++ ) {

             final Entity entity;

             try {
                 entityMap.put( "key", i+1 );
                 entity = em.create(itemType, entityMap);
             }
             catch ( Exception ex ) {
                 throw new RuntimeException( "Error creating entity", ex );
             }

             entityRefs.add( new SimpleEntityRef( entity.getType(), entity.getUuid() ) );
             if ( (i+1) % 10 == 0 ) {
                 logger.info( "Created {} entities", i+1 );
             }
         }


         app.waitForQueueDrainAndRefreshIndex(10000);

         long timeFirstPutDone = System.currentTimeMillis();
         logger.info("Finished adding first lot of entities at {}", timeFirstPutDone);

         try {
             //Wait to make sure that the time on the next entry changes
             Thread.sleep(2000);
         }
         catch (Exception e) {
         }

         for (int i = 0; i < ENTITIES_TO_ADD_AFTER_TIME; i++) {

             final Entity entity;

             try {
                 entityMap.put("key", ENTITIES_TO_DELETE + i+1);
                 entity = em.create(itemType, entityMap);
             } catch (Exception ex) {
                 throw new RuntimeException("Error creating entity", ex);
             }

             entityRefs.add(new SimpleEntityRef(entity.getType(), entity.getUuid()));
             logger.info("Created {} entities after delete time with key {} and uuid {} at {} ", i + 1, entity.getProperty("key"), entity.getUuid(), entity.getCreated());

         }


         app.waitForQueueDrainAndRefreshIndex(15000);

         final CollectionDeleteRequestBuilder builder =
             collectionDeleteService.getBuilder()
                 .withApplicationId( em.getApplicationId() )
                 .withCollection(collectionName)
                 .withEndTimestamp(timeFirstPutDone);

         CollectionDeleteService.CollectionDeleteStatus status = collectionDeleteService.deleteCollection(builder);

         assertNotNull( status.getJobId(), "JobId is present" );

         logger.info( "Delete collection" );


         waitForDelete( status, collectionDeleteService );

         //app.waitForQueueDrainAndRefreshIndex(15000);

         // ----------------- test that we can read the entries after the timestamp

         retryReadData( em, collectionName, ENTITIES_TO_ADD_AFTER_TIME, 60);
     }

     /**
      * Wait for the delete to occur
      */
     private void waitForDelete( final CollectionDeleteService.CollectionDeleteStatus status, final CollectionDeleteService collectionDeleteService )
         throws InterruptedException, IllegalArgumentException {
         if (status != null) {
             logger.info("waitForDelete: jobID={}", status.getJobId());
         } else {
             logger.info("waitForDelete: error, status = null");
             throw new IllegalArgumentException("collectionDeleteStatus = null");
         }
         while ( true ) {

             try {
                 final CollectionDeleteService.CollectionDeleteStatus updatedStatus =
                     collectionDeleteService.getStatus( status.getJobId() );

                 if (updatedStatus == null) {
                     logger.info("waitForDelete: updated status is null");
                 } else {
                     logger.info("waitForDelete: status={} numberProcessed={}",
                         updatedStatus.getStatus().toString(), updatedStatus.getNumberProcessed());

                     if ( updatedStatus.getStatus() == CollectionDeleteService.Status.COMPLETE ) {
                         break;
                     }
                 }
             }
             catch ( IllegalArgumentException iae ) {
                 //swallow.  Thrown if our job can't be found.  I.E hasn't updated yet
             }


             Thread.sleep( 1000 );
         }
     }

     private int retryReadData(EntityManager em, String collectionName, int expectedEntities,  int retry) throws Exception {
         int count = -1;
         Set<Entity> uniqueRemEnts = new HashSet<Entity>();
         do {
             try {
                 count = readData(em, collectionName, expectedEntities, uniqueRemEnts);
             } catch (Exception ignore) {
                 logger.info( "caught exception ", ignore);
             }
             logger.info( "read {} expected {}" , count, expectedEntities);
         } while (count != expectedEntities && --retry >=0);
         assertEquals( "Did not get expected entities", expectedEntities, count );
         return count;
     }

     private int readData(EntityManager em, String collectionName, int expectedEntities, Set<Entity> uniqueRemEnts)
         throws Exception {

         app.waitForQueueDrainAndRefreshIndex();

         Results results = em.getCollection(em.getApplicationRef(), collectionName, null, expectedEntities,
             Query.Level.ALL_PROPERTIES, false);


         while ( true ) {

             if (results.getEntities().size() == 0) {
                 break;
             }


             UUID lastEntityUUID = null;
             for ( Entity e : results.getEntities() ) {

                 assertEquals(2000, e.getProperty("key2"));

                 if (uniqueRemEnts.size() % 100 == 0) {
                     logger.info("read {} entities", uniqueRemEnts.size());
                 }
                 lastEntityUUID = e.getUuid();
                 uniqueRemEnts.add(e);
                 logger.info("Found remaining entity {} with key {}", lastEntityUUID, e.getProperty("key"));
             }

             results = em.getCollection(em.getApplicationRef(), collectionName, lastEntityUUID, expectedEntities,
                 Query.Level.ALL_PROPERTIES, false);

         }

         if (uniqueRemEnts.size() != expectedEntities) {
             logger.info("Expected {} did not match actual {}", expectedEntities, uniqueRemEnts.size());
             if (uniqueRemEnts.size() < 20) {
                 for (Entity e : uniqueRemEnts) {
                     Object key = e.getProperty("key");
                     logger.info("Entity key {} uuid {} created {}", key,e.getUuid(), e.getCreated());
                 }
             }
         }

         assertEquals( "Did not get expected entities", expectedEntities, uniqueRemEnts.size() );
         return uniqueRemEnts.size();
     }

     private int countEntities( EntityManager em, String collectionName, int expectedEntities)
            throws Exception {

            app.waitForQueueDrainAndRefreshIndex();

            Query q = Query.fromQL( "select * where key1=1000" ).withLimit( 1000 );
            Results results = em.searchCollectionConsistent( em.getApplicationRef(), collectionName, q, expectedEntities );

            int count = 0;
            while ( true ) {

                count += results.size();


                if ( results.hasCursor() ) {
                    logger.info( "Counted {} : query again with cursor", count );
                    q.setCursor( results.getCursor() );
                    results = em.searchCollection( em.getApplicationRef(), collectionName, q );
                }
                else {
                    break;
                }
            }

            assertEquals( "Did not get expected entities", expectedEntities, count );
            return count;
        }


 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one or more
	* contributor license agreements. See the NOTICE file distributed with
	* this work for additional information regarding copyright ownership.
	* The ASF licenses this file to You 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.apache.usergrid.persistence;


	import static org.junit.Assert.assertEquals;
	import static org.junit.Assert.assertNotNull;

	import java.util.ArrayList;
	import java.util.HashMap;
	import java.util.HashSet;
	import java.util.List;
	import java.util.Map;
	import java.util.Set;
	import java.util.UUID;

	import org.apache.commons.lang.RandomStringUtils;
	import org.apache.usergrid.AbstractCoreIT;
	import org.apache.usergrid.corepersistence.index.CollectionDeleteRequestBuilder;
	import org.apache.usergrid.corepersistence.index.CollectionDeleteService;
	import org.junit.After;
	import org.junit.Before;
	import org.junit.Test;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;




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



	private static final int ENTITIES_TO_DELETE = 1100;
	private static final int ENTITIES_TO_ADD_AFTER_TIME = 5;


	@Before
	public void startReporting() {

	if (logger.isDebugEnabled()) {
	logger.debug("Starting metrics reporting");
	}
	}


	@After
	public void printReport() {
	logger.debug( "Printing metrics report" );
	}


	@Test( timeout = 240000 )
	public void clearOneCollection() throws Exception {

	logger.info( "Started clearOneCollection()" );

	String rand = RandomStringUtils.randomAlphanumeric( 5 );
	final UUID appId = setup.createApplication( "org_" + rand, "app_" + rand );

	final EntityManager em = setup.getEmf().getEntityManager( appId );

	final CollectionDeleteService collectionDeleteService = setup.getInjector().getInstance( CollectionDeleteService.class );

	// ----------------- create a bunch of entities

	Map<String, Object> entityMap = new HashMap<String, Object>() {{
	put( "key1", 1000 );
	put( "key2", 2000 );
	put( "key3", "Some value" );
	}};

	String collectionName = "items";
	String itemType = "item";


	List<EntityRef> entityRefs = new ArrayList<EntityRef>();
	for ( int i = 0; i < ENTITIES_TO_DELETE; i++ ) {

	final Entity entity;

	try {
	entityMap.put( "key", i+1 );
	entity = em.create(itemType, entityMap);
	}
	catch ( Exception ex ) {
	throw new RuntimeException( "Error creating entity", ex );
	}

	entityRefs.add( new SimpleEntityRef( entity.getType(), entity.getUuid() ) );
	if ( (i+1) % 10 == 0 ) {
	logger.info( "Created {} entities", i+1 );
	}
	}


	app.waitForQueueDrainAndRefreshIndex(10000);

	long timeFirstPutDone = System.currentTimeMillis();
	logger.info("Finished adding first lot of entities at {}", timeFirstPutDone);

	try {
	//Wait to make sure that the time on the next entry changes
	Thread.sleep(2000);
	}
	catch (Exception e) {
	}

	for (int i = 0; i < ENTITIES_TO_ADD_AFTER_TIME; i++) {

	final Entity entity;

	try {
	entityMap.put("key", ENTITIES_TO_DELETE + i+1);
	entity = em.create(itemType, entityMap);
	} catch (Exception ex) {
	throw new RuntimeException("Error creating entity", ex);
	}

	entityRefs.add(new SimpleEntityRef(entity.getType(), entity.getUuid()));
	logger.info("Created {} entities after delete time with key {} and uuid {} at {} ", i + 1, entity.getProperty("key"), entity.getUuid(), entity.getCreated());

	}


	app.waitForQueueDrainAndRefreshIndex(15000);

	final CollectionDeleteRequestBuilder builder =
	collectionDeleteService.getBuilder()
	.withApplicationId( em.getApplicationId() )
	.withCollection(collectionName)
	.withEndTimestamp(timeFirstPutDone);

	CollectionDeleteService.CollectionDeleteStatus status = collectionDeleteService.deleteCollection(builder);

	assertNotNull( status.getJobId(), "JobId is present" );

	logger.info( "Delete collection" );


	waitForDelete( status, collectionDeleteService );

	//app.waitForQueueDrainAndRefreshIndex(15000);

	// ----------------- test that we can read the entries after the timestamp

	retryReadData( em, collectionName, ENTITIES_TO_ADD_AFTER_TIME, 60);
	}

	/**
	* Wait for the delete to occur
	*/
	private void waitForDelete( final CollectionDeleteService.CollectionDeleteStatus status, final CollectionDeleteService collectionDeleteService )
	throws InterruptedException, IllegalArgumentException {
	if (status != null) {
	logger.info("waitForDelete: jobID={}", status.getJobId());
	} else {
	logger.info("waitForDelete: error, status = null");
	throw new IllegalArgumentException("collectionDeleteStatus = null");
	}
	while ( true ) {

	try {
	final CollectionDeleteService.CollectionDeleteStatus updatedStatus =
	collectionDeleteService.getStatus( status.getJobId() );

	if (updatedStatus == null) {
	logger.info("waitForDelete: updated status is null");
	} else {
	logger.info("waitForDelete: status={} numberProcessed={}",
	updatedStatus.getStatus().toString(), updatedStatus.getNumberProcessed());

	if ( updatedStatus.getStatus() == CollectionDeleteService.Status.COMPLETE ) {
	break;
	}
	}
	}
	catch ( IllegalArgumentException iae ) {
	//swallow. Thrown if our job can't be found. I.E hasn't updated yet
	}


	Thread.sleep( 1000 );
	}
	}

	private int retryReadData(EntityManager em, String collectionName, int expectedEntities, int retry) throws Exception {
	int count = -1;
	Set<Entity> uniqueRemEnts = new HashSet<Entity>();
	do {
	try {
	count = readData(em, collectionName, expectedEntities, uniqueRemEnts);
	} catch (Exception ignore) {
	logger.info( "caught exception ", ignore);
	}
	logger.info( "read {} expected {}" , count, expectedEntities);
	} while (count != expectedEntities && --retry >=0);
	assertEquals( "Did not get expected entities", expectedEntities, count );
	return count;
	}

	private int readData(EntityManager em, String collectionName, int expectedEntities, Set<Entity> uniqueRemEnts)
	throws Exception {

	app.waitForQueueDrainAndRefreshIndex();

	Results results = em.getCollection(em.getApplicationRef(), collectionName, null, expectedEntities,
	Query.Level.ALL_PROPERTIES, false);


	while ( true ) {

	if (results.getEntities().size() == 0) {
	break;
	}


	UUID lastEntityUUID = null;
	for ( Entity e : results.getEntities() ) {

	assertEquals(2000, e.getProperty("key2"));

	if (uniqueRemEnts.size() % 100 == 0) {
	logger.info("read {} entities", uniqueRemEnts.size());
	}
	lastEntityUUID = e.getUuid();
	uniqueRemEnts.add(e);
	logger.info("Found remaining entity {} with key {}", lastEntityUUID, e.getProperty("key"));
	}

	results = em.getCollection(em.getApplicationRef(), collectionName, lastEntityUUID, expectedEntities,
	Query.Level.ALL_PROPERTIES, false);

	}

	if (uniqueRemEnts.size() != expectedEntities) {
	logger.info("Expected {} did not match actual {}", expectedEntities, uniqueRemEnts.size());
	if (uniqueRemEnts.size() < 20) {
	for (Entity e : uniqueRemEnts) {
	Object key = e.getProperty("key");
	logger.info("Entity key {} uuid {} created {}", key,e.getUuid(), e.getCreated());
	}
	}
	}

	assertEquals( "Did not get expected entities", expectedEntities, uniqueRemEnts.size() );
	return uniqueRemEnts.size();
	}

	private int countEntities( EntityManager em, String collectionName, int expectedEntities)
	throws Exception {

	app.waitForQueueDrainAndRefreshIndex();

	Query q = Query.fromQL( "select * where key1=1000" ).withLimit( 1000 );
	Results results = em.searchCollectionConsistent( em.getApplicationRef(), collectionName, q, expectedEntities );

	int count = 0;
	while ( true ) {

	count += results.size();


	if ( results.hasCursor() ) {
	logger.info( "Counted {} : query again with cursor", count );
	q.setCursor( results.getCursor() );
	results = em.searchCollection( em.getApplicationRef(), collectionName, q );
	}
	else {
	break;
	}
	}

	assertEquals( "Did not get expected entities", expectedEntities, count );
	return count;
	}


	}