maytasm commented on a change in pull request #9644: Add missing integration tests for the compaction by the coordinator URL: https://github.com/apache/druid/pull/9644#discussion_r407037142
########## File path: integration-tests/src/test/java/org/apache/druid/tests/coordinator/duty/ITAutoCompactionTest.java ########## @@ -0,0 +1,351 @@ +/* + * 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.druid.tests.coordinator.duty; + +import com.google.inject.Inject; +import org.apache.commons.io.IOUtils; +import org.apache.druid.indexer.partitions.SecondaryPartitionType; +import org.apache.druid.java.util.common.ISE; +import org.apache.druid.java.util.common.StringUtils; +import org.apache.druid.java.util.common.logger.Logger; +import org.apache.druid.server.coordinator.CoordinatorCompactionConfig; +import org.apache.druid.server.coordinator.DataSourceCompactionConfig; +import org.apache.druid.testing.IntegrationTestingConfig; +import org.apache.druid.testing.clients.CompactionResourceTestClient; +import org.apache.druid.testing.guice.DruidTestModuleFactory; +import org.apache.druid.testing.utils.ITRetryUtil; +import org.apache.druid.tests.TestNGGroup; +import org.apache.druid.tests.indexer.AbstractITBatchIndexTest; +import org.apache.druid.tests.indexer.AbstractIndexerTest; +import org.apache.druid.timeline.DataSegment; +import org.joda.time.Period; +import org.testng.Assert; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Guice; +import org.testng.annotations.Test; + +import java.io.Closeable; +import java.io.IOException; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.List; +import java.util.UUID; + +@Test(groups = {TestNGGroup.OTHER_INDEX}) +@Guice(moduleFactory = DruidTestModuleFactory.class) +public class ITAutoCompactionTest extends AbstractIndexerTest +{ + private static final Logger LOG = new Logger(ITAutoCompactionTest.class); + private static final String INDEX_TASK = "/indexer/wikipedia_index_task.json"; + private static final String INDEX_QUERIES_RESOURCE = "/indexer/wikipedia_index_queries.json"; + private static final int MAX_ROWS_PER_SEGMENT_COMPACTED = 10000; + private static final Period SKIP_OFFSET_FROM_LATEST = Period.seconds(0); + + @Inject + protected CompactionResourceTestClient compactionResource; + + @Inject + private IntegrationTestingConfig config; + + private String fullDatasourceName; + + @BeforeMethod + public void setup() throws Exception + { + // Set comapction slot to 10 + updateCompactionTaskSlot(0.5, 10); + fullDatasourceName = "wikipedia_index_test_" + UUID.randomUUID() + config.getExtraDatasourceNameSuffix(); + } + + @Test + public void testAutoCompactionDutySubmitAndVerifyCompaction() throws Exception + { + loadData(INDEX_TASK); + try (final Closeable ignored = unloader(fullDatasourceName)) { + final List<String> intervalsBeforeCompaction = coordinator.getSegmentIntervals(fullDatasourceName); + intervalsBeforeCompaction.sort(null); + // 4 segments across 2 days (4 total)... + verifySegmentsCount(4); + verifyQuery(INDEX_QUERIES_RESOURCE); + + submitCompactionConfig(MAX_ROWS_PER_SEGMENT_COMPACTED, Period.days(1)); + forceTriggerAutoCompaction(); + //...compacted into 1 new segment for 1 day. 1 day compacted and 1 day skipped/remains uncompacted. (5 total) + verifySegmentsCount(5); + verifyQuery(INDEX_QUERIES_RESOURCE); + verifySegmentsCompacted(1, MAX_ROWS_PER_SEGMENT_COMPACTED); + checkCompactionIntervals(intervalsBeforeCompaction); + + submitCompactionConfig(MAX_ROWS_PER_SEGMENT_COMPACTED, SKIP_OFFSET_FROM_LATEST); + forceTriggerAutoCompaction(); + //...compacted into 1 new segment for the remaining one day. 2 day compacted and 0 day uncompacted. (6 total) + verifySegmentsCount(6); + verifyQuery(INDEX_QUERIES_RESOURCE); + verifySegmentsCompacted(2, MAX_ROWS_PER_SEGMENT_COMPACTED); + checkCompactionIntervals(intervalsBeforeCompaction); + } + } + + @Test + public void testAutoCompactionDutyCanUpdateCompactionConfig() throws Exception + { + loadData(INDEX_TASK); + try (final Closeable ignored = unloader(fullDatasourceName)) { + final List<String> intervalsBeforeCompaction = coordinator.getSegmentIntervals(fullDatasourceName); + intervalsBeforeCompaction.sort(null); + // 4 segments across 2 days (4 total)... + verifySegmentsCount(4); + verifyQuery(INDEX_QUERIES_RESOURCE); + + // Dummy compaction config which will be overwritten + submitCompactionConfig(10000, SKIP_OFFSET_FROM_LATEST); + // New compaction config should overwrites the existing compaction config + submitCompactionConfig(1, SKIP_OFFSET_FROM_LATEST); + forceTriggerAutoCompaction(); + + // Instead of merging segments, the updated config will split segments! + //...compacted into 10 new segments across 2 days. 5 new segments each day (14 total) + verifySegmentsCount(14); + verifyQuery(INDEX_QUERIES_RESOURCE); + verifySegmentsCompacted(10, 2); + + checkCompactionIntervals(intervalsBeforeCompaction); + } + } + + @Test + public void testAutoCompactionDutyCanDeleteCompactionConfig() throws Exception + { + loadData(INDEX_TASK); + try (final Closeable ignored = unloader(fullDatasourceName)) { + final List<String> intervalsBeforeCompaction = coordinator.getSegmentIntervals(fullDatasourceName); + intervalsBeforeCompaction.sort(null); + // 4 segments across 2 days (4 total)... + verifySegmentsCount(4); + verifyQuery(INDEX_QUERIES_RESOURCE); + + submitCompactionConfig(MAX_ROWS_PER_SEGMENT_COMPACTED, SKIP_OFFSET_FROM_LATEST); + deleteCompactionConfig(); + forceTriggerAutoCompaction(); + + // ...should remains unchanged (4 total) + verifySegmentsCount(4); + verifyQuery(INDEX_QUERIES_RESOURCE); + verifySegmentsCompacted(0, null); + + checkCompactionIntervals(intervalsBeforeCompaction); + } + } + + @Test + public void testAutoCompactionDutyCanUpdateTaskSlots() throws Exception + { + loadData(INDEX_TASK); + try (final Closeable ignored = unloader(fullDatasourceName)) { + final List<String> intervalsBeforeCompaction = coordinator.getSegmentIntervals(fullDatasourceName); + intervalsBeforeCompaction.sort(null); + // 4 segments across 2 days (4 total)... + verifySegmentsCount(4); + verifyQuery(INDEX_QUERIES_RESOURCE); + + // Skips first day. Should only compact one out of two days. + submitCompactionConfig(MAX_ROWS_PER_SEGMENT_COMPACTED, SKIP_OFFSET_FROM_LATEST); + + // Set compactionTaskSlotRatio to 0 to prevent any compaction + updateCompactionTaskSlot(0, 100); + forceTriggerAutoCompaction(); + // ...should remains unchanged (4 total) + verifySegmentsCount(4); + verifyQuery(INDEX_QUERIES_RESOURCE); + verifySegmentsCompacted(0, null); + checkCompactionIntervals(intervalsBeforeCompaction); + + // Set maxCompactionTaskSlots to 0 to prevent any compaction + updateCompactionTaskSlot(0.1, 0); + forceTriggerAutoCompaction(); + // ...should remains unchanged (4 total) + verifySegmentsCount(4); + verifyQuery(INDEX_QUERIES_RESOURCE); + verifySegmentsCompacted(0, null); + checkCompactionIntervals(intervalsBeforeCompaction); + + // Update compaction slots to be 1 + updateCompactionTaskSlot(1, 1); + forceTriggerAutoCompaction(); + // One day compacted (1 new segment) and one day remains uncompacted. (5 total) + verifySegmentsCount(5); + verifyQuery(INDEX_QUERIES_RESOURCE); + verifySegmentsCompacted(1, MAX_ROWS_PER_SEGMENT_COMPACTED); + checkCompactionIntervals(intervalsBeforeCompaction); + Assert.assertEquals(compactionResource.getCompactionProgress(fullDatasourceName).get("remainingSegmentSize"), "14312"); + // Run compaction again to compact the remaining day + forceTriggerAutoCompaction(); + // Remaining day compacted (1 new segment). Now both days compacted (6 total) + verifySegmentsCount(6); + verifyQuery(INDEX_QUERIES_RESOURCE); + verifySegmentsCompacted(2, MAX_ROWS_PER_SEGMENT_COMPACTED); + checkCompactionIntervals(intervalsBeforeCompaction); +// Assert.assertEquals(compactionResource.getCompactionProgress(fullDatasourceName).get("remainingSegmentSize"), "0"); Review comment: @jihoonson Not sure if this is expected but when I call the API for get compaction progress here I got status[400 Bad Request] content[{"error":"unknown dataSource"}] instead of 0. I expected 0 since all segments of the datasource have been compacted. ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@druid.apache.org For additional commands, e-mail: commits-h...@druid.apache.org
