forked from apache/accumulo
-
Notifications
You must be signed in to change notification settings - Fork 2
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Adds property to minor compact based on age of data in memory
Adds a new property that will initiate a minor compaction based on the age of data in memory. Tablets now roughly track that age of the first write to a tablets in memory map and when this age exceeds a configured threshold a minor compaction is initiated. fixes apache#3397
- Loading branch information
1 parent
dcf951f
commit 8cb0982
Showing
6 changed files
with
256 additions
and
10 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
166 changes: 166 additions & 0 deletions
166
test/src/main/java/org/apache/accumulo/test/ScanServerMaxLatencyIT.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,166 @@ | ||
| /* | ||
| * 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 | ||
| * | ||
| * https://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.accumulo.test; | ||
|
|
||
| import static org.apache.accumulo.core.client.ScannerBase.ConsistencyLevel.EVENTUAL; | ||
| import static org.apache.accumulo.core.client.ScannerBase.ConsistencyLevel.IMMEDIATE; | ||
| import static org.junit.jupiter.api.Assertions.assertEquals; | ||
| import static org.junit.jupiter.api.Assertions.assertTrue; | ||
|
|
||
| import java.security.SecureRandom; | ||
| import java.util.ArrayList; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
| import java.util.concurrent.Callable; | ||
| import java.util.concurrent.ExecutorService; | ||
| import java.util.concurrent.Executors; | ||
| import java.util.concurrent.TimeUnit; | ||
|
|
||
| import org.apache.accumulo.core.client.Accumulo; | ||
| import org.apache.accumulo.core.client.AccumuloClient; | ||
| import org.apache.accumulo.core.client.ScannerBase; | ||
| import org.apache.accumulo.core.client.admin.NewTableConfiguration; | ||
| import org.apache.accumulo.core.conf.Property; | ||
| import org.apache.accumulo.core.data.Mutation; | ||
| import org.apache.accumulo.core.util.Timer; | ||
| import org.apache.accumulo.minicluster.ServerType; | ||
| import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl; | ||
| import org.apache.accumulo.test.functional.ConfigurableMacBase; | ||
| import org.apache.accumulo.test.util.Wait; | ||
| import org.apache.hadoop.conf.Configuration; | ||
| import org.apache.hadoop.io.Text; | ||
| import org.junit.jupiter.api.Test; | ||
|
|
||
| public class ScanServerMaxLatencyIT extends ConfigurableMacBase { | ||
|
|
||
| protected void configure(MiniAccumuloConfigImpl cfg, Configuration hadoopCoreSite) { | ||
| cfg.setProperty(Property.SSERV_CACHED_TABLET_METADATA_EXPIRATION, "2s"); | ||
| } | ||
|
|
||
| @Test | ||
| public void testMaxLatency() throws Exception { | ||
| final String[] tables = this.getUniqueNames(3); | ||
| final String table1 = tables[0]; | ||
| final String table2 = tables[1]; | ||
| final String table3 = tables[2]; | ||
|
|
||
| getCluster().getConfig().setNumScanServers(1); | ||
| getCluster().getClusterControl().startAllServers(ServerType.SCAN_SERVER); | ||
|
|
||
| ExecutorService executor = Executors.newCachedThreadPool(); | ||
| try (var client = Accumulo.newClient().from(getClientProperties()).build()) { | ||
|
|
||
| Wait.waitFor(() -> !client.instanceOperations().getScanServers().isEmpty()); | ||
|
|
||
| var ntc = new NewTableConfiguration(); | ||
| ntc.setProperties(Map.of(Property.TABLE_MINC_COMPACT_MAXAGE.getKey(), "2s")); | ||
| client.tableOperations().create(table1, ntc); | ||
| client.tableOperations().create(table2); | ||
| ntc = new NewTableConfiguration(); | ||
| ntc.setProperties(Map.of(Property.TABLE_MINC_COMPACT_IDLETIME.getKey(), "2s")); | ||
| client.tableOperations().create(table3, ntc); | ||
|
|
||
| Timer timer = Timer.startNew(); | ||
|
|
||
| executor.submit(createWriterTask(client, table1, timer)); | ||
| executor.submit(createWriterTask(client, table2, timer)); | ||
|
|
||
| long lastMaxSeen = -1; | ||
| int changes = 0; | ||
|
|
||
| List<Long> deltas = new ArrayList<>(); | ||
|
|
||
| while (changes < 4) { | ||
| Thread.sleep(250); | ||
| var currElapsed = timer.elapsed(TimeUnit.MILLISECONDS); | ||
| var maxElapsedInTable = readMaxElapsed(client, EVENTUAL, table1); | ||
|
|
||
| if (maxElapsedInTable > 0 && maxElapsedInTable != lastMaxSeen) { | ||
| log.info("new max elapsed seen {} {}", lastMaxSeen, maxElapsedInTable); | ||
| changes++; | ||
| lastMaxSeen = maxElapsedInTable; | ||
| } | ||
|
|
||
| if (maxElapsedInTable > 0) { | ||
| // This is difference in elapsed time written to the table vs the most recent elapsed | ||
| // time. | ||
| deltas.add(currElapsed - maxElapsedInTable); | ||
| } | ||
|
|
||
| // The other table does not have the setting to minor compact based on age, so should never | ||
| // see any data for it from the scan server. | ||
| assertEquals(-1, readMaxElapsed(client, EVENTUAL, table2)); | ||
| // The background thread is writing to this table every 100ms so it should not be considered | ||
| // idle and therefor should not minor compact. | ||
| assertEquals(-1, readMaxElapsed(client, EVENTUAL, table3)); | ||
| } | ||
|
|
||
| var stats = deltas.stream().mapToLong(l -> l).summaryStatistics(); | ||
| log.info("Delta stats : {}", stats); | ||
| // Should usually see data within 4 seconds, but not always because the timings config are | ||
| // when things should start to happen and not when they are guaranteed to finish. Would expect | ||
| // the average to be less than 4 seconds and the max less than 8 seconds. These numbers may | ||
| // not hold if running test on a heavily loaded machine. | ||
| assertTrue(stats.getAverage() > 500 && stats.getAverage() < 4000); | ||
| assertTrue(stats.getMax() < 8000); | ||
| assertTrue(stats.getCount() > 9); | ||
|
|
||
| executor.shutdownNow(); | ||
| executor.awaitTermination(600, TimeUnit.SECONDS); | ||
|
|
||
| assertEquals(-1, readMaxElapsed(client, EVENTUAL, table2)); | ||
| // This test assumes the 2nd table returns nothing because it is reading it via scan server. | ||
| // Validate this test assumption by doing an immediate scan using tablet server which should | ||
| // return data. | ||
| assertTrue(readMaxElapsed(client, IMMEDIATE, table2) > 0); | ||
| // Now that nothing is writing its expected that max read by an immediate scan will see any | ||
| // data an eventual scan would see. | ||
| assertTrue( | ||
| readMaxElapsed(client, IMMEDIATE, table1) >= readMaxElapsed(client, EVENTUAL, table1)); | ||
| } | ||
|
|
||
| } | ||
|
|
||
| private long readMaxElapsed(AccumuloClient client, ScannerBase.ConsistencyLevel consistency, | ||
| String table) throws Exception { | ||
| try (var scanner = client.createScanner(table)) { | ||
| scanner.setConsistencyLevel(consistency); | ||
| scanner.fetchColumn(new Text("elapsed"), new Text("nanos")); | ||
| return scanner.stream().mapToLong(e -> Long.parseLong(e.getValue().toString(), 10)).max() | ||
| .orElse(-1); | ||
| } | ||
| } | ||
|
|
||
| private static Callable<Void> createWriterTask(AccumuloClient client, String table, Timer timer) { | ||
| SecureRandom random = new SecureRandom(); | ||
| Callable<Void> writerTask = () -> { | ||
| try (var writer = client.createBatchWriter(table)) { | ||
| while (true) { | ||
| var elapsed = timer.elapsed(TimeUnit.MILLISECONDS); | ||
| Mutation m = new Mutation(Long.toHexString(random.nextLong())); | ||
| m.put("elapsed", "nanos", "" + elapsed); | ||
| writer.addMutation(m); | ||
| writer.flush(); | ||
| Thread.sleep(100); | ||
| } | ||
| } | ||
| }; | ||
| return writerTask; | ||
| } | ||
| } |