You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by jo...@apache.org on 2026/08/28 22:13:13 UTC
(impala) branch master updated: IMPALA-14618: Fix EventProcessor race with catalog reset
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
The following commit(s) were added to refs/heads/master by this push:
new 6a7effcbd IMPALA-14618: Fix EventProcessor race with catalog reset
6a7effcbd is described below
commit 6a7effcbde4eab32cecafe64a0e156cb97abd375
Author: stiga-huang <hu...@gmail.com>
AuthorDate: Fri Jul 24 11:46:25 2026 +0800
IMPALA-14618: Fix EventProcessor race with catalog reset
Race of CREATE_DATABASE event
-----------------------------
While processing a CREATE_DATABASE event, the check-and-add in
CatalogOpExecutor#addDbIfNotRemovedLater() is not protected by the
catalog version lock so has race with the concurrent catalog reset. Here
is the sequence when things go wrong:
thread1: catalog reset starts
thread2: CREATE_DATABASE event finds the db doesn't exist, i.e., getDb()
returns null.
thread1: catalog reset loads the db with the latest table list
thread2: CREATE_DATABASE overwrites the Db instance with a new one that
has an empty table list. The table list is expected to be updated by
following CREATE_TABLE events.
thread1: finishes catalog reset. Update last synced event id and restart
EventProcessor to start from it. The CREATE_TABLE events are
skipped. Leaving the db with an empty table list.
Race of CREATE_TABLE event
-----------------------------
While processing a CREATE_TABLE event, a concurrent reset could finish
and restarts event processor to polling events from a new event id,
which skips some events. If there is a corresponding DROP_TABLE event
not being polled and processed, the CREATE_TABLE event will add a stale
table in the db. Note that the table is expected to be removed by the
DROP_TABLE event which is skipped.
Event processor modes
---------------------
When hierarchical event processing is disabled, these are not problems
since catalog resets can't start concurrently when an HMS event is being
processing. This is ensured by the instance lock of
MetastoreEventsProcessor:
- Catalog resets invoke MetastoreEventsProcessor.pause() at the
begining. It's a synchronized method.
- Processing an event in the main thread of MetastoreEventsProcessor is
guarded by "synchronized (this)".
When hierarchical event processing is enabled, the main thread of
MetastoreEventsProcessor just dispatch events. Events are processed in
other threads so can run concurrently with catalog resets.
Solution
--------
This patch fixes the CREATE_DATABASE issue by moving the check-and-set
codes in CatalogOpExecutor#addDbIfNotRemovedLater() into
CatalogServiceCatalog so they can be wrapped within
WriteLockAndLookupDb. This fixes the race with concurrent catalog reset.
DROP_DATABASE event also has the same issue in
CatalogOpExecutor#removeDbIfNotAddedLater() that the check-and-remove
operations are not protected by the catalog version lock. This patch
also moves them into CatalogServiceCatalog to be wrapped with
WriteLockAndLookupDb.
There is an issue that the createEventId of dbs are not set during
catalog reset, which impacts the skipping logic of DROP_DATABASE events.
This patch also fixes this issue.
The issue of applying stale CREATE_TABLE event is fixed by correctly
skipping the event:
- If the corresponding DROP_TABLE event has been polled from HMS, add
it to the DeleteEventLog when queuing it so the thread processing
CREATE_TABLE event knows the table is removed later.
- If the DROP_TABLE event is not in DeleteEventLog, we fallback to
check the createEventId of the db which will be updated to the latest
event id at which reset starts. If the CREATE_TABLE event id is
smaller than this id, the event can be skipped.
Testing
- Added e2e tests by adding debug actions to inject sleeps.
- Ran test_failover_catchup_timeout_and_reset and
TestEventProcessingRace 100 times.
Assisted-by: Opus 4.8 (Claude Code)
Change-Id: Ifa0833c09cd7ec2461a0f2d1db7ab5a72ab6822b
Reviewed-on: http://gerrit.cloudera.org:8080/24629
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
.../impala/catalog/CatalogServiceCatalog.java | 53 +++++++++++++++++++-
.../impala/catalog/events/MetastoreEvents.java | 6 ---
.../impala/catalog/events/TableEventExecutor.java | 9 ++++
.../apache/impala/service/CatalogOpExecutor.java | 50 ++++++++-----------
.../java/org/apache/impala/util/DebugUtils.java | 9 ++++
tests/custom_cluster/test_events_custom_configs.py | 57 ++++++++++++++++++++++
6 files changed, 146 insertions(+), 38 deletions(-)
diff --git a/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java b/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
index 89d887b47..87518bb0b 100644
--- a/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
+++ b/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
@@ -1444,7 +1444,7 @@ public class CatalogServiceCatalog extends Catalog {
}
/**
- * Utillity class to facilitate simple read or write locking of versionLock_ and
+ * Utility class to facilitate simple read or write locking of versionLock_ and
* lookup the requested database from dbCache_. The lock is released when close()
* is called, which is automatically called when using try-with-resources statement.
* Use either the ReadLockAndLookupDb or WriteLockAndLookupDb subclass to get
@@ -2354,6 +2354,7 @@ public class CatalogServiceCatalog extends Catalog {
loadFunctions(newDb, prefetchedObjects.getNativeFunctions(),
prefetchedObjects.getJavaFunctions());
newDb.setCatalogVersion(incrementAndGetCatalogVersion());
+ newDb.setCreateEventId(currentEventId);
int numTables = 0;
List<TTableName> tblsToBackgroundLoad = new ArrayList<>();
@@ -2721,6 +2722,56 @@ public class CatalogServiceCatalog extends Catalog {
return removedDb;
}
+ /**
+ * Atomically adds a database only if it does not already exist, after waiting for any
+ * ongoing reset() to finish loading it.
+ * @return the newly added Db, or null if the db already exists in the cache.
+ */
+ public @Nullable Db addDbIfNotExists(String dbName,
+ org.apache.hadoop.hive.metastore.api.Database msDb, long eventId) {
+ try (WriteLockAndLookupDb lookup = new WriteLockAndLookupDb(dbName)) {
+ if (lookup.getDb() != null) {
+ // The db already exists, possibly just rebuilt by a concurrent reset(). Do not
+ // overwrite it (which would drop its table list).
+ return null;
+ }
+ Db newDb = new Db(dbName, msDb);
+ newDb.setCatalogVersion(incrementAndGetCatalogVersion());
+ newDb.setCreateEventId(eventId);
+ addDb(newDb);
+ return newDb;
+ }
+ }
+
+ /**
+ * Atomically removes a database only if it exists and its createEventId is not newer
+ * than 'eventId'.
+ * @return true if the db was removed
+ */
+ public boolean removeDbIfNotAddedLater(String dbName, long eventId) {
+ try (WriteLockAndLookupDb lookup = new WriteLockAndLookupDb(dbName)) {
+ Db db = lookup.getDb();
+ if (db == null) {
+ LOG.info(
+ "EventId: {} Skipping the event since database {} does not exist anymore",
+ eventId, dbName);
+ return false;
+ }
+ // if this database has been created after this drop database event is generated
+ // the createdEventId of the database will be higher than eventId. In such case
+ // it means that catalog has recreated this database again and events processor
+ // is just receiving the earlier drop database event. We should ignore such event.
+ if (db.getCreateEventId() > eventId) {
+ LOG.info(
+ "EventId: {} Not removing the database {} since the create event id is {}",
+ eventId, dbName, db.getCreateEventId());
+ return false;
+ }
+ removeDbLocked(dbName);
+ return true;
+ }
+ }
+
/**
* Helper function to clean up the state associated with a removed database. It creates
* the entries in the delete log for 'db' as well as for its tables and functions
diff --git a/fe/src/main/java/org/apache/impala/catalog/events/MetastoreEvents.java b/fe/src/main/java/org/apache/impala/catalog/events/MetastoreEvents.java
index 2d7f774e6..3c496e088 100644
--- a/fe/src/main/java/org/apache/impala/catalog/events/MetastoreEvents.java
+++ b/fe/src/main/java/org/apache/impala/catalog/events/MetastoreEvents.java
@@ -1863,12 +1863,6 @@ public class MetastoreEvents {
// Don't put the event processor into error state, instead ignore this event.
errorLog("Create table {}.{} failed because the database does not exist cache." +
" Ignoring the CREATE_TABLE event", dbName_, tblName_, ex);
- } catch (CatalogException e) {
- // if a DatabaseNotFoundException is caught here it means either we incorrectly
- // determined that the event needs to be processed instead of skipped, or we
- // somehow missed the previous create database event.
- throw new MetastoreNotificationException(
- debugString("Unable to process event"), e);
}
}
diff --git a/fe/src/main/java/org/apache/impala/catalog/events/TableEventExecutor.java b/fe/src/main/java/org/apache/impala/catalog/events/TableEventExecutor.java
index d3363a9fe..fa57dd4a9 100644
--- a/fe/src/main/java/org/apache/impala/catalog/events/TableEventExecutor.java
+++ b/fe/src/main/java/org/apache/impala/catalog/events/TableEventExecutor.java
@@ -191,6 +191,13 @@ public class TableEventExecutor {
((DbBarrierEvent) event).incrExpectedProceedCount();
} else if (event.getEventType() == MetastoreEventType.DROP_TABLE) {
skipEventId_.set(event.getEventId());
+ // Also add this to DeleteEventLog so CREATE_TABLE event that is already
+ // in-processing can be skipped (see CatalogOpExecutor#addTableIfNotRemovedLater).
+ // This is a best-effort. The DROP_TABLE or ALTER_TABLE rename events could still
+ // not reach here, e.g. not polled from HMS or still at dispatcher/DBEventExecutor
+ // thread.
+ eventProcessor.getDeleteEventLog().addRemovedObject(event.getEventId(),
+ DeleteEventLog.getTblKey(event.getDbName(), event.getTableName()));
}
synchronized (processorLock_) {
Preconditions.checkState(!isTerminating());
@@ -253,6 +260,8 @@ public class TableEventExecutor {
Counter eventSkipCounter = event.getMetrics()
.getCounter(MetastoreEventsProcessor.EVENTS_SKIPPED_METRIC);
eventSkipCounter.inc();
+ event.debugLog("Skipped since table is dropped later by event {}",
+ skipEventId_.get());
event.debugLog("Incremented skipped metric to {}", eventSkipCounter.getCount());
return true;
}
diff --git a/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java b/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
index e19b5e1a5..cc761c340 100644
--- a/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
+++ b/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
@@ -889,6 +889,8 @@ public class CatalogOpExecutor {
getMetastoreDdlLock().lock();
try {
String dbName = msTbl.getDbName();
+ DebugUtils.executeDebugAction(BackendConfig.INSTANCE.debugActions(),
+ DebugUtils.GET_DB_DELAY);
Db db = catalog_.getDb(dbName);
DeleteEventLog deleteEventLog = catalog_.getMetastoreEventProcessor()
.getDeleteEventLog();
@@ -909,11 +911,7 @@ public class CatalogOpExecutor {
if (existingTable != null) {
LOG.debug("EventId: {} Table {} was not added since "
+ "it already exists in catalog.", eventId, existingTable.getFullName());
- if (existingTable.getCreateEventId() != eventId) {
- LOG.warn("Existing table {} create event Id: {} does not match the "
- + "event id: {}", existingTable.getFullName(),
- existingTable.getCreateEventId(), eventId);
- }
+ existingTable.setCreateEventId(eventId, false);
return false;
}
// table does not exist in catalog. We must make sure that the table was
@@ -923,6 +921,15 @@ public class CatalogOpExecutor {
"EventId: {} Table was not added since it was removed later", eventId);
return false;
}
+ // DeleteEventLog might miss the corresponding DROP_TABLE event, e.g. when the event
+ // hasn't been pulled from HMS but a catalog reset moves the start event id forward.
+ // In such cases, we check createEventId of the db to see if current event is stale.
+ if (eventId <= db.getCreateEventId()) {
+ LOG.debug("EventId: {} Table {}.{} was not added since db is re-created or "
+ + "re-loaded later at event {}",
+ eventId, dbName, tblName, db.getCreateEventId());
+ return false;
+ }
// set the createEventId of the table to eventId since we are adding table
// due to the given eventId.
Table incompleteTable = IncompleteTable.createUninitializedTable(db, tblName,
@@ -1011,16 +1018,15 @@ public class CatalogOpExecutor {
getMetastoreDdlLock().lock();
try {
String dbName = msDb.getName();
- Db db = catalog_.getDb(dbName);
DeleteEventLog deleteEventLog = catalog_.getMetastoreEventProcessor()
.getDeleteEventLog();
- if (db == null) {
- if (!deleteEventLog.wasRemovedAfter(eventId, DeleteEventLog.getKey(msDb))) {
- catalog_.addDb(dbName, msDb, eventId);
- return true;
- }
+ if (deleteEventLog.wasRemovedAfter(eventId, DeleteEventLog.getKey(msDb))) {
+ return false;
}
- return false;
+ LOG.info("EventId: {} Processing CREATE_DATABASE for db: {}", eventId, dbName);
+ DebugUtils.executeDebugAction(BackendConfig.INSTANCE.debugActions(),
+ DebugUtils.ADD_DB_AFTER_CHECK_DELAY);
+ return catalog_.addDbIfNotExists(dbName, msDb, eventId) != null;
} finally {
getMetastoreDdlLock().unlock();
}
@@ -1036,25 +1042,7 @@ public class CatalogOpExecutor {
public boolean removeDbIfNotAddedLater(long eventId, String dbName) {
getMetastoreDdlLock().lock();
try {
- Db catalogDb = catalog_.getDb(dbName);
- if (catalogDb == null) {
- LOG.info(
- "EventId: {} Skipping the event since database {} does not exist anymore",
- eventId, dbName);
- return false;
- }
- // if this database has been created after this drop database event is generated
- // the createdEventId of the database will be higher than eventId. In such case
- // if means that catalog has recreated this database again and events processor
- // is just receiving the earlier drop database event. We should ignore such event.
- if (catalogDb.getCreateEventId() > eventId) {
- LOG.info(
- "EventId: {} Not removing the database {} since the create event id is {}",
- eventId, dbName, catalogDb.getCreateEventId());
- return false;
- }
- catalog_.removeDb(dbName);
- return true;
+ return catalog_.removeDbIfNotAddedLater(dbName, eventId);
} finally {
getMetastoreDdlLock().unlock();
}
diff --git a/fe/src/main/java/org/apache/impala/util/DebugUtils.java b/fe/src/main/java/org/apache/impala/util/DebugUtils.java
index b5d402dba..4f9a7f6ed 100644
--- a/fe/src/main/java/org/apache/impala/util/DebugUtils.java
+++ b/fe/src/main/java/org/apache/impala/util/DebugUtils.java
@@ -84,6 +84,15 @@ public class DebugUtils {
// debug action label to inject a delay in processing each HMS event
public static final String EVENT_PROCESSING_DELAY = "catalogd_event_processing_delay";
+ // debug action label to inject a delay in addDbIfNotRemovedLater after the delete-log
+ // check and before adding the db to catalog.
+ public static final String ADD_DB_AFTER_CHECK_DELAY =
+ "catalogd_add_db_after_check_delay";
+
+ // debug action label to inject a delay before getting the db from catalog.
+ // Currently only used in addTableIfNotRemovedLater.
+ public static final String GET_DB_DELAY = "catalogd_get_db_delay";
+
// debug action label for introducing delay in table metadata loading by catalogd.
public static final String LOAD_METADATA_DELAY = "catalogd_load_metadata_delay";
diff --git a/tests/custom_cluster/test_events_custom_configs.py b/tests/custom_cluster/test_events_custom_configs.py
index 671bcacab..4c7cdb93a 100644
--- a/tests/custom_cluster/test_events_custom_configs.py
+++ b/tests/custom_cluster/test_events_custom_configs.py
@@ -2358,3 +2358,60 @@ class TestEventProcessingCatchupModeCustom(TestEventProcessingCustomConfigsBase)
self.assert_catalogd_log_contains("INFO", log_regex, expected_count=1, timeout_s=20)
log_regex = r"ALTER_TABLE.*{}.* Catch-up Mode: Skipping".format(tbl)
self.assert_catalogd_log_contains("INFO", log_regex, expected_count=1, timeout_s=20)
+
+
+class TestEventProcessingRace(TestEventProcessingCustomConfigsBase):
+ """Regression test for IMPALA-14618: processing HMS events racing with a catalog reset()
+ (INVALIDATE METADATA / catalogd-HA failover) must not leave the database with a
+ stale table list."""
+
+ @CustomClusterTestSuite.with_args(
+ catalogd_args="--debug_actions=catalogd_add_db_after_check_delay:SLEEP@3000",
+ disable_log_buffering=True)
+ def test_create_db_table_survives_reset(self, unique_name):
+ """Create a db + table via Hive. While the CREATE_DATABASE event is parked, issue
+ INVALIDATE METADATA to trigger a reset that rebuilds the db with its table list. After
+ both complete, the table must still be present."""
+ db = unique_name + "_db"
+ self.run_stmt_in_hive("drop database if exists {0} cascade".format(db))
+ self.execute_query("invalidate metadata")
+ try:
+ self.run_stmt_in_hive("create database {0}".format(db))
+ self.run_stmt_in_hive("create table {0}.t1 (id int)".format(db))
+
+ # Wait until the CREATE_DATABASE event is being processed.
+ self.assert_catalogd_log_contains(
+ "INFO", r"Sleeping for .* to execute debug action")
+
+ # Trigger a full reset while the CREATE_DATABASE event is parked. It finishes
+ # before the CREATE_DATABASE event is processed/skipped.
+ self.execute_query("invalidate metadata")
+
+ # The CREATE_DATABASE event should be skipped.
+ self.assert_catalogd_log_contains("INFO", r"Database {0} was not added".format(db))
+
+ tables = self.execute_query("show tables in {0}".format(db)).data
+ assert "t1" in tables
+ self.execute_query_expect_success(self.client, "describe {0}.t1".format(db))
+ finally:
+ self.run_stmt_in_hive("drop database if exists {0} cascade".format(db))
+
+ @CustomClusterTestSuite.with_args(
+ catalogd_args="--debug_actions=catalogd_get_db_delay:SLEEP@6000",
+ disable_log_buffering=True)
+ def test_create_drop_table_with_reset(self, unique_database):
+ EventProcessorUtils.wait_for_event_processing(self)
+ self.run_stmt_in_hive("create table {0}.tbl (id int)".format(unique_database))
+ self.assert_catalogd_log_contains(
+ "INFO", r"Sleeping for .* to execute debug action")
+
+ # Drop the table and run a concurrent reset to load the latest table list.
+ # These happen before the event processing thread wakes up.
+ self.run_stmt_in_hive("drop table {0}.tbl".format(unique_database))
+ self.execute_query("invalidate metadata")
+
+ # The CREATE_TABLE event should be skipped.
+ self.assert_catalogd_log_contains(
+ "INFO", r"Table {0}.tbl was not added".format(unique_database))
+ tables = self.execute_query("show tables in {0}".format(unique_database)).data
+ assert len(tables) == 0