| /* |
| * 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.cassandra.db.compaction; |
| |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.UUID; |
| |
| import com.google.common.collect.Lists; |
| import org.junit.Assert; |
| import org.junit.Test; |
| |
| import org.apache.cassandra.db.lifecycle.SSTableSet; |
| import org.apache.cassandra.io.sstable.format.SSTableReader; |
| import org.apache.cassandra.repair.consistent.LocalSessionAccessor; |
| import org.apache.cassandra.utils.FBUtilities; |
| import org.apache.cassandra.utils.UUIDGen; |
| |
| public class PendingRepairManagerTest extends AbstractPendingRepairTest |
| { |
| /** |
| * If a local session is ongoing, it should not be cleaned up |
| */ |
| @Test |
| public void needsCleanupInProgress() |
| { |
| PendingRepairManager prm = csm.getPendingRepairManagers().get(0); |
| |
| UUID repairID = registerSession(cfs, true, true); |
| LocalSessionAccessor.prepareUnsafe(repairID, COORDINATOR, PARTICIPANTS); |
| SSTableReader sstable = makeSSTable(true); |
| mutateRepaired(sstable, repairID, false); |
| prm.addSSTable(sstable); |
| Assert.assertNotNull(prm.get(repairID)); |
| |
| Assert.assertFalse(prm.canCleanup(repairID)); |
| } |
| |
| /** |
| * If a local session is finalized, it should be cleaned up |
| */ |
| @Test |
| public void needsCleanupFinalized() |
| { |
| PendingRepairManager prm = csm.getPendingRepairManagers().get(0); |
| |
| UUID repairID = registerSession(cfs, true, true); |
| LocalSessionAccessor.prepareUnsafe(repairID, COORDINATOR, PARTICIPANTS); |
| SSTableReader sstable = makeSSTable(true); |
| mutateRepaired(sstable, repairID, false); |
| prm.addSSTable(sstable); |
| Assert.assertNotNull(prm.get(repairID)); |
| LocalSessionAccessor.finalizeUnsafe(repairID); |
| |
| Assert.assertTrue(prm.canCleanup(repairID)); |
| } |
| |
| /** |
| * If a local session has failed, it should be cleaned up |
| */ |
| @Test |
| public void needsCleanupFailed() |
| { |
| PendingRepairManager prm = csm.getPendingRepairManagers().get(0); |
| |
| UUID repairID = registerSession(cfs, true, true); |
| LocalSessionAccessor.prepareUnsafe(repairID, COORDINATOR, PARTICIPANTS); |
| SSTableReader sstable = makeSSTable(true); |
| mutateRepaired(sstable, repairID, false); |
| prm.addSSTable(sstable); |
| Assert.assertNotNull(prm.get(repairID)); |
| LocalSessionAccessor.failUnsafe(repairID); |
| |
| Assert.assertTrue(prm.canCleanup(repairID)); |
| } |
| |
| @Test |
| public void needsCleanupNoSession() |
| { |
| UUID fakeID = UUIDGen.getTimeUUID(); |
| PendingRepairManager prm = new PendingRepairManager(cfs, null, false); |
| Assert.assertTrue(prm.canCleanup(fakeID)); |
| } |
| |
| @Test |
| public void estimateRemainingTasksInProgress() |
| { |
| PendingRepairManager prm = csm.getPendingRepairManagers().get(0); |
| |
| UUID repairID = registerSession(cfs, true, true); |
| LocalSessionAccessor.prepareUnsafe(repairID, COORDINATOR, PARTICIPANTS); |
| SSTableReader sstable = makeSSTable(true); |
| mutateRepaired(sstable, repairID, false); |
| prm.addSSTable(sstable); |
| Assert.assertNotNull(prm.get(repairID)); |
| |
| Assert.assertEquals(0, prm.getEstimatedRemainingTasks()); |
| Assert.assertEquals(0, prm.getNumPendingRepairFinishedTasks()); |
| } |
| |
| @Test |
| public void estimateRemainingFinishedRepairTasks() |
| { |
| PendingRepairManager prm = csm.getPendingRepairManagers().get(0); |
| |
| UUID repairID = registerSession(cfs, true, true); |
| LocalSessionAccessor.prepareUnsafe(repairID, COORDINATOR, PARTICIPANTS); |
| SSTableReader sstable = makeSSTable(true); |
| mutateRepaired(sstable, repairID, false); |
| prm.addSSTable(sstable); |
| Assert.assertNotNull(prm.get(repairID)); |
| Assert.assertNotNull(prm.get(repairID)); |
| LocalSessionAccessor.finalizeUnsafe(repairID); |
| |
| Assert.assertEquals(0, prm.getEstimatedRemainingTasks()); |
| Assert.assertEquals(1, prm.getNumPendingRepairFinishedTasks()); |
| } |
| |
| @Test |
| public void getNextBackgroundTask() |
| { |
| PendingRepairManager prm = csm.getPendingRepairManagers().get(0); |
| |
| UUID repairID = registerSession(cfs, true, true); |
| LocalSessionAccessor.prepareUnsafe(repairID, COORDINATOR, PARTICIPANTS); |
| SSTableReader sstable = makeSSTable(true); |
| mutateRepaired(sstable, repairID, false); |
| prm.addSSTable(sstable); |
| |
| repairID = registerSession(cfs, true, true); |
| LocalSessionAccessor.prepareUnsafe(repairID, COORDINATOR, PARTICIPANTS); |
| sstable = makeSSTable(true); |
| mutateRepaired(sstable, repairID, false); |
| prm.addSSTable(sstable); |
| LocalSessionAccessor.finalizeUnsafe(repairID); |
| |
| Assert.assertEquals(2, prm.getSessions().size()); |
| Assert.assertNull(prm.getNextBackgroundTask(FBUtilities.nowInSeconds())); |
| AbstractCompactionTask compactionTask = prm.getNextRepairFinishedTask(); |
| try |
| { |
| Assert.assertNotNull(compactionTask); |
| Assert.assertSame(PendingRepairManager.RepairFinishedCompactionTask.class, compactionTask.getClass()); |
| PendingRepairManager.RepairFinishedCompactionTask cleanupTask = (PendingRepairManager.RepairFinishedCompactionTask) compactionTask; |
| Assert.assertEquals(repairID, cleanupTask.getSessionID()); |
| } |
| finally |
| { |
| compactionTask.transaction.abort(); |
| } |
| } |
| |
| @Test |
| public void getNextBackgroundTaskNoSessions() |
| { |
| PendingRepairManager prm = csm.getPendingRepairManagers().get(0); |
| Assert.assertNull(prm.getNextBackgroundTask(FBUtilities.nowInSeconds())); |
| } |
| |
| /** |
| * If all sessions should be cleaned up, getNextBackgroundTask should return null |
| */ |
| @Test |
| public void getNextBackgroundTaskAllCleanup() throws Exception |
| { |
| PendingRepairManager prm = csm.getPendingRepairManagers().get(0); |
| UUID repairID = registerSession(cfs, true, true); |
| LocalSessionAccessor.prepareUnsafe(repairID, COORDINATOR, PARTICIPANTS); |
| |
| SSTableReader sstable = makeSSTable(true); |
| mutateRepaired(sstable, repairID, false); |
| prm.addSSTable(sstable); |
| Assert.assertNotNull(prm.get(repairID)); |
| Assert.assertNotNull(prm.get(repairID)); |
| LocalSessionAccessor.finalizeUnsafe(repairID); |
| |
| Assert.assertNull(prm.getNextBackgroundTask(FBUtilities.nowInSeconds())); |
| |
| } |
| |
| @Test |
| public void maximalTaskNeedsCleanup() |
| { |
| PendingRepairManager prm = csm.getPendingRepairManagers().get(0); |
| |
| UUID repairID = registerSession(cfs, true, true); |
| LocalSessionAccessor.prepareUnsafe(repairID, COORDINATOR, PARTICIPANTS); |
| SSTableReader sstable = makeSSTable(true); |
| mutateRepaired(sstable, repairID, false); |
| prm.addSSTable(sstable); |
| Assert.assertNotNull(prm.get(repairID)); |
| Assert.assertNotNull(prm.get(repairID)); |
| LocalSessionAccessor.finalizeUnsafe(repairID); |
| |
| Collection<AbstractCompactionTask> tasks = prm.getMaximalTasks(FBUtilities.nowInSeconds(), false); |
| try |
| { |
| Assert.assertEquals(1, tasks.size()); |
| } |
| finally |
| { |
| tasks.stream().forEach(t -> t.transaction.abort()); |
| } |
| } |
| |
| @Test |
| public void userDefinedTaskTest() |
| { |
| PendingRepairManager prm = csm.getPendingRepairManagers().get(0); |
| UUID repairId = registerSession(cfs, true, true); |
| SSTableReader sstable = makeSSTable(true); |
| mutateRepaired(sstable, repairId, false); |
| prm.addSSTable(sstable); |
| |
| try (CompactionTasks tasks = csm.getUserDefinedTasks(Collections.singleton(sstable), 100)) |
| { |
| Assert.assertEquals(1, tasks.size()); |
| } |
| } |
| |
| @Test |
| public void mixedPendingSessionsTest() |
| { |
| PendingRepairManager prm = csm.getPendingRepairManagers().get(0); |
| UUID repairId = registerSession(cfs, true, true); |
| UUID repairId2 = registerSession(cfs, true, true); |
| SSTableReader sstable = makeSSTable(true); |
| SSTableReader sstable2 = makeSSTable(true); |
| |
| mutateRepaired(sstable, repairId, false); |
| mutateRepaired(sstable2, repairId2, false); |
| prm.addSSTable(sstable); |
| prm.addSSTable(sstable2); |
| try (CompactionTasks tasks = csm.getUserDefinedTasks(Lists.newArrayList(sstable, sstable2), 100)) |
| { |
| Assert.assertEquals(2, tasks.size()); |
| } |
| } |
| |
| /** |
| * Tests that a IllegalSSTableArgumentException is thrown if we try to get |
| * scanners for an sstable that isn't pending repair |
| */ |
| @Test(expected = PendingRepairManager.IllegalSSTableArgumentException.class) |
| public void getScannersInvalidSSTable() throws Exception |
| { |
| PendingRepairManager prm = csm.getPendingRepairManagers().get(0); |
| SSTableReader sstable = makeSSTable(true); |
| prm.getScanners(Collections.singleton(sstable), Collections.singleton(RANGE1)); |
| } |
| |
| /** |
| * Tests that a IllegalSSTableArgumentException is thrown if we try to get |
| * scanners for an sstable that isn't pending repair |
| */ |
| @Test(expected = PendingRepairManager.IllegalSSTableArgumentException.class) |
| public void getOrCreateInvalidSSTable() throws Exception |
| { |
| PendingRepairManager prm = csm.getPendingRepairManagers().get(0); |
| SSTableReader sstable = makeSSTable(true); |
| prm.getOrCreate(sstable); |
| } |
| |
| @Test |
| public void sessionHasData() |
| { |
| PendingRepairManager prm = csm.getPendingRepairManagers().get(0); |
| |
| UUID repairID = registerSession(cfs, true, true); |
| LocalSessionAccessor.prepareUnsafe(repairID, COORDINATOR, PARTICIPANTS); |
| |
| Assert.assertFalse(prm.hasDataForSession(repairID)); |
| SSTableReader sstable = makeSSTable(true); |
| mutateRepaired(sstable, repairID, false); |
| prm.addSSTable(sstable); |
| Assert.assertTrue(prm.hasDataForSession(repairID)); |
| } |
| |
| @Test |
| public void noEmptyCompactionTask() |
| { |
| PendingRepairManager prm = csm.getPendingRepairManagers().get(0); |
| SSTableReader sstable = makeSSTable(false); |
| UUID id = UUID.randomUUID(); |
| mutateRepaired(sstable, id, false); |
| prm.getOrCreate(sstable); |
| cfs.truncateBlocking(); |
| Assert.assertFalse(cfs.getSSTables(SSTableSet.LIVE).iterator().hasNext()); |
| Assert.assertNull(cfs.getCompactionStrategyManager().getNextBackgroundTask(0)); |
| |
| } |
| } |