diff --git common/src/java/org/apache/hadoop/hive/conf/HiveConf.java common/src/java/org/apache/hadoop/hive/conf/HiveConf.java index 6678de6..2d2f20f 100644 --- common/src/java/org/apache/hadoop/hive/conf/HiveConf.java +++ common/src/java/org/apache/hadoop/hive/conf/HiveConf.java @@ -1570,7 +1570,7 @@ public void setSparkConfigUpdated(boolean isSparkConfigUpdated) { "compaction re-queued."), HIVE_COMPACTOR_CHECK_INTERVAL("hive.compactor.check.interval", "300s", - new TimeValidator(TimeUnit.SECONDS), + new TimeValidator(TimeUnit.SECONDS, 30L, true, 60*60*24*7L, true), "Time in seconds between checks to see if any tables or partitions need to be\n" + "compacted. This should be kept high because each check for compaction requires\n" + "many calls against the NameNode.\n" + diff --git common/src/java/org/apache/hadoop/hive/conf/Validator.java common/src/java/org/apache/hadoop/hive/conf/Validator.java index 3fb09b9..9b503a6 100644 --- common/src/java/org/apache/hadoop/hive/conf/Validator.java +++ common/src/java/org/apache/hadoop/hive/conf/Validator.java @@ -242,10 +242,21 @@ public TimeUnit getTimeUnit() { return timeUnit; } + public Long getMin(TimeUnit toTimeUnit) { + return min == null ? null : HiveConf.toTime(Long.toString(min), timeUnit, toTimeUnit); + } + public Long getMax(TimeUnit toTimeUnit) { + return max == null ? null : HiveConf.toTime(Long.toString(max), timeUnit, toTimeUnit); + } + + /** + * @param value must have time unit suffix + * @return + */ @Override public String validate(String value) { try { - long time = HiveConf.toTime(value, timeUnit, timeUnit); + long time = HiveConf.toTime(value, null, timeUnit); if (min != null && (minInclusive ? time < min : time <= min)) { return value + " is smaller than " + timeString(min); } @@ -258,6 +269,20 @@ public String validate(String value) { return null; } + /** + * returns a value adjusted to fall withing valid range. + * @param value must have time unit suffix + */ + public Long getValidValue(String value, TimeUnit toTimeUnit) { + long time = HiveConf.toTime(value, null, timeUnit); + if (min != null && (minInclusive ? time < min : time <= min)) { + return HiveConf.toTime(Long.toString(min), timeUnit, toTimeUnit); + } + if (max != null && (maxInclusive ? time > max : time >= max)) { + return HiveConf.toTime(Long.toString(max), timeUnit, toTimeUnit); + } + return HiveConf.toTime(value, timeUnit, toTimeUnit); + } public String toDescription() { String description = "Expects a time value with unit " + diff --git ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Initiator.java ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Initiator.java index 3705a34..31c99b2 100644 --- ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Initiator.java +++ ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Initiator.java @@ -17,12 +17,15 @@ */ package org.apache.hadoop.hive.ql.txn.compactor; +import com.google.common.annotations.VisibleForTesting; + import org.apache.hadoop.fs.FileStatus; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.hive.common.FileUtils; import org.apache.hadoop.hive.common.ValidTxnList; import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.conf.Validator; import org.apache.hadoop.hive.metastore.api.CompactionRequest; import org.apache.hadoop.hive.metastore.api.CompactionType; import org.apache.hadoop.hive.metastore.api.MetaException; @@ -169,10 +172,25 @@ public void run() { @Override public void init(AtomicBoolean stop, AtomicBoolean looped) throws MetaException { super.init(stop, looped); - checkInterval = - conf.getTimeVar(HiveConf.ConfVars.HIVE_COMPACTOR_CHECK_INTERVAL, TimeUnit.MILLISECONDS) ; + long rawConfig = conf.getTimeVar(HiveConf.ConfVars.HIVE_COMPACTOR_CHECK_INTERVAL, TimeUnit.MILLISECONDS); + Validator.TimeValidator v = (Validator.TimeValidator) HiveConf.ConfVars.HIVE_COMPACTOR_CHECK_INTERVAL.getValidator(); + String errorMsg = v.validate(rawConfig + "ms"); + if(errorMsg == null) { + checkInterval = rawConfig; + return; + } + checkInterval = v.getValidValue(rawConfig + "ms", TimeUnit.MILLISECONDS); + LOG.warn(HiveConf.ConfVars.HIVE_COMPACTOR_CHECK_INTERVAL.name() + " value is out of range: " + + errorMsg + " Will use: " + checkInterval + "ms (Unless " + HiveConf.ConfVars.HIVE_IN_TEST.name() + " is set)"); + if(conf.getBoolVar(HiveConf.ConfVars.HIVE_IN_TEST)) { + //in test mode, we want to use whatever the user set + checkInterval = HiveConf.toTime(Long.toString(rawConfig), TimeUnit.MILLISECONDS, TimeUnit.MILLISECONDS); + } + } + @VisibleForTesting + long getCheckInterval() { + return checkInterval; } - private void recoverFailedCompactions(boolean remoteOnly) throws MetaException { if (!remoteOnly) txnHandler.revokeFromLocalWorkers(Worker.hostname()); txnHandler.revokeTimedoutWorkers(HiveConf.getTimeVar(conf, diff --git ql/src/test/org/apache/hadoop/hive/ql/txn/compactor/TestInitiator.java ql/src/test/org/apache/hadoop/hive/ql/txn/compactor/TestInitiator.java index a31e2d1..de056c8 100644 --- ql/src/test/org/apache/hadoop/hive/ql/txn/compactor/TestInitiator.java +++ ql/src/test/org/apache/hadoop/hive/ql/txn/compactor/TestInitiator.java @@ -18,6 +18,7 @@ package org.apache.hadoop.hive.ql.txn.compactor; import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.conf.Validator; import org.apache.hadoop.hive.metastore.api.AbortTxnRequest; import org.apache.hadoop.hive.metastore.api.CommitTxnRequest; import org.apache.hadoop.hive.metastore.api.CompactionRequest; @@ -45,6 +46,7 @@ import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; /** * Tests for the compactor Initiator thread. @@ -63,6 +65,21 @@ public void nothing() throws Exception { // survival test. startInitiator(); } + @Test + public void testConfig() throws Exception { + conf.setTimeVar(HiveConf.ConfVars.HIVE_COMPACTOR_CHECK_INTERVAL, 1, TimeUnit.SECONDS); + Validator.TimeValidator v = (Validator.TimeValidator)HiveConf.ConfVars.HIVE_COMPACTOR_CHECK_INTERVAL.getValidator(); + Long validValue = v.getValidValue( + conf.getTimeVar(HiveConf.ConfVars.HIVE_COMPACTOR_CHECK_INTERVAL, TimeUnit.MILLISECONDS) + "ms", + TimeUnit.MILLISECONDS); + Assert.assertEquals("Expected in-range value", v.getMin(TimeUnit.MILLISECONDS), validValue); + + conf.setBoolVar(HiveConf.ConfVars.HIVE_IN_TEST, true); + Initiator initiator = new Initiator(); + initiator.setHiveConf(conf); + initiator.init(new AtomicBoolean(true), new AtomicBoolean(false)); + Assert.assertEquals("HIVE_IN_TEST override didn't work", 1000, initiator.getCheckInterval()); + } @Test public void recoverFailedLocalWorkers() throws Exception {