This is an automated email from the ASF dual-hosted git repository.
abstractdog pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hive.git
The following commit(s) were added to refs/heads/master by this push:
new 459f25c826a HIVE-30024: Fix scheduled queries for HA configuration
(#6766)
459f25c826a is described below
commit 459f25c826a8f489045d62ae0403c2f741707411
Author: InvisibleProgrammer <[email protected]>
AuthorDate: Tue Sep 22 11:04:56 2026 +0200
HIVE-30024: Fix scheduled queries for HA configuration (#6766)
---
.../apache/hadoop/hive/metastore/ObjectStore.java | 2 +-
.../client/TestMetastoreScheduledQueries.java | 114 ++++++++++++++++++---
2 files changed, 98 insertions(+), 18 deletions(-)
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ObjectStore.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ObjectStore.java
index 2dffb7feacb..3ca67b9f6b1 100644
---
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ObjectStore.java
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ObjectStore.java
@@ -2953,7 +2953,7 @@ private void
processScheduledQueryPolicies(ScheduledQueryProgressInfo info) thro
query = pm.newQuery(MScheduledExecution.class);
query.setFilter("scheduledQuery == currentSchedule");
- query.setOrdering("scheduledExecutionId descending");
+ query.setOrdering("endTime descending, scheduledExecutionId descending");
query.declareParameters("MScheduledQuery currentSchedule");
query.setRange(0, lastN);
List<MScheduledExecution> list = (List<MScheduledExecution>)
query.execute(schq);
diff --git
a/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/client/TestMetastoreScheduledQueries.java
b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/client/TestMetastoreScheduledQueries.java
index 1ac58ded074..a212d20b4ec 100644
---
a/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/client/TestMetastoreScheduledQueries.java
+++
b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/client/TestMetastoreScheduledQueries.java
@@ -23,6 +23,7 @@
import static org.apache.hadoop.hive.metastore.utils.TestTxnDbUtil.prepDb;
import static
org.apache.hadoop.hive.metastore.utils.TestTxnDbUtil.queryToString;
import static
org.apache.hadoop.hive.metastore.utils.TestTxnDbUtil.setConfValues;
+import static org.awaitility.Awaitility.await;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertFalse;
@@ -31,8 +32,8 @@
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
-import java.sql.SQLException;
import java.sql.Statement;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
@@ -41,11 +42,13 @@
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import javax.jdo.JDOObjectNotFoundException;
import javax.jdo.PersistenceManager;
import javax.jdo.Query;
+import javax.jdo.Transaction;
import org.apache.curator.shaded.com.google.common.collect.Lists;
import org.apache.hadoop.conf.Configuration;
@@ -103,11 +106,7 @@ public TestMetastoreScheduledQueries(String name,
AbstractMetaStoreService metaS
@Before
public void setUp() throws Exception {
-
metaStore.getConf().set(MetastoreConf.ConfVars.SCHEDULED_QUERIES_AUTODISABLE_COUNT.getVarname(),
"-1");
-
metaStore.getConf().set(MetastoreConf.ConfVars.SCHEDULED_QUERIES_SKIP_OPPORTUNITIES_AFTER_FAILURES.getVarname(),
- "0");
- client = metaStore.getClient();
-
+ prepareMetastoreClientWithAutoDisable("-1", "0");
}
@After
@@ -345,28 +344,22 @@ public ScheduledQueryPollResponse call() throws Exception
{
@Test
public void testDisable1() throws Exception {
-
metaStore.getConf().set(MetastoreConf.ConfVars.SCHEDULED_QUERIES_AUTODISABLE_COUNT.getVarname(),
"1");
- client.close();
- client = metaStore.getClient();
+ prepareMetastoreClientWithAutoDisable("1", null);
+
testDisableInternal(2, 5, "dis1");
}
@Test
public void testDisable2() throws Exception {
-
metaStore.getConf().set(MetastoreConf.ConfVars.SCHEDULED_QUERIES_AUTODISABLE_COUNT.getVarname(),
"2");
- client.close();
- client = metaStore.getClient();
+ prepareMetastoreClientWithAutoDisable("2", null);
+
testDisableInternal(3, 5, "dis2");
}
@Test
public void testSkip2() throws Exception {
-
metaStore.getConf().set(MetastoreConf.ConfVars.SCHEDULED_QUERIES_AUTODISABLE_COUNT.getVarname(),
"4");
-
metaStore.getConf().set(MetastoreConf.ConfVars.SCHEDULED_QUERIES_SKIP_OPPORTUNITIES_AFTER_FAILURES.getVarname(),
- "2");
- client.close();
- client = metaStore.getClient();
+ prepareMetastoreClientWithAutoDisable("4", "2");
testDisableInternal(5, 6, "skip2");
try (PersistenceManager pm =
PersistenceManagerProvider.getPersistenceManager()) {
@@ -395,6 +388,93 @@ public void testSkip2() throws Exception {
}
}
+
+ /**
+ * This test stimulates DataNucleus' id pre-allocation feature and creates
id gaps between executions
+ * The test checks if auto disabling still works assuming the execution id
generation doesn't generate strictly
+ * monotonous growing values
+ *
+ * @throws Exception
+ */
+ @Test
+ public void testAutoDisablePolicyUsesEndTime() throws Exception {
+ String testNamespace = "endTimeNamespace";
+ prepareMetastoreClientWithAutoDisable("2", "0");
+
+ ScheduledQueryKey key = new ScheduledQueryKey("q1", testNamespace);
+ createEverySecondSchq(key);
+
+ long exec1Id = pollAndGetExecutionId(testNamespace);
+ client.scheduledQueryProgress(
+ new ScheduledQueryProgressInfo(exec1Id, QueryState.FINISHED,
"executor-query-id"));
+
+ long exec2Id = pollAndGetExecutionId(testNamespace);
+ ScheduledQueryProgressInfo exec2Info =
+ new ScheduledQueryProgressInfo(exec2Id, QueryState.FAILED,
"executor-query-id");
+ exec2Info.setErrorMessage("some issue happened");
+ client.scheduledQueryProgress(exec2Info);
+
+ stimulateIdPreallocation(exec1Id, exec2Id);
+
+ long exec3Id = pollAndGetExecutionId(testNamespace);
+ ScheduledQueryProgressInfo exec3Info =
+ new ScheduledQueryProgressInfo(exec3Id, QueryState.FAILED,
"executor-query-id");
+ exec2Info.setErrorMessage("some issue happened");
+ client.scheduledQueryProgress(exec3Info);
+
+ ScheduledQuery schq = client.getScheduledQuery(key);
+ assertTrue("Scheduled query should be enabled", schq.isEnabled());
+ }
+
+ private void stimulateIdPreallocation(long exec1Id, long exec2Id) {
+ int skewedRecentEndTime = getEpochSeconds();
+ try (PersistenceManager pm =
PersistenceManagerProvider.getPersistenceManager()) {
+ Transaction tx = pm.currentTransaction();
+ tx.begin();
+ MScheduledExecution exec1 = pm.getObjectById(MScheduledExecution.class,
exec1Id);
+ exec1.setEndTime(skewedRecentEndTime);
+ MScheduledExecution exec2 = pm.getObjectById(MScheduledExecution.class,
exec2Id);
+ exec2.setEndTime(1);
+ tx.commit();
+ }
+ }
+
+ private void prepareMetastoreClientWithAutoDisable(String autoDisableCount,
String skipOpportunitiesAfterFailures) throws MetaException {
+ if (autoDisableCount != null) {
+
metaStore.getConf().set(ConfVars.SCHEDULED_QUERIES_AUTODISABLE_COUNT.getVarname(),
autoDisableCount);
+ }
+
+ if (skipOpportunitiesAfterFailures != null) {
+
metaStore.getConf().set(ConfVars.SCHEDULED_QUERIES_SKIP_OPPORTUNITIES_AFTER_FAILURES.getVarname(),
skipOpportunitiesAfterFailures);
+ }
+
+ if (client != null) {
+ client.close();
+ }
+ client = metaStore.getClient();
+ }
+
+ private long pollAndGetExecutionId(String testNamespace) {
+ ScheduledQueryPollRequest request = new
ScheduledQueryPollRequest(testNamespace);
+ AtomicReference<Long> executionId = new AtomicReference<>();
+
+ await()
+ .atMost(Duration.ofSeconds(3))
+ .pollInterval(Duration.ofMillis(100))
+ .until(() -> {
+ ScheduledQueryPollResponse response =
client.scheduledQueryPoll(request);
+
+ if (response.isSetQuery()) {
+ executionId.set(response.getExecutionId());
+ return true;
+ }
+
+ return false;
+ });
+
+ return executionId.get();
+ }
+
/**
* Simulates some schq failure scenario.
*
