You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by la...@apache.org on 2023/06/07 20:04:26 UTC
[impala] 03/03: IMPALA-12189: updateCatalog should handle failures in createTblTransaction
This is an automated email from the ASF dual-hosted git repository.
laszlog pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 58590376ed3a1b2cac2bbc3b6a59d5cd34c53672
Author: stiga-huang <hu...@gmail.com>
AuthorDate: Wed Jun 7 13:09:48 2023 +0800
IMPALA-12189: updateCatalog should handle failures in createTblTransaction
updateCatalog() invokes createTblTransaction() for transactional tables.
It's called after acquiring the table lock. The write lock of catalog's
versionLock will also be acquired by the current thread. Whenever we hit
an exception, we should release those locks. This patch moves the code
calling createTblTransaction() into the exception handling scope.
Tests:
- Add a debug action to abort the transaction in updateCatalog() so
createTblTransaction() will fail.
- Add e2e test for the error handling.
Change-Id: I3a64764d0568fc1e6c6f4c52f9e220df3130bd84
Reviewed-on: http://gerrit.cloudera.org:8080/20020
Reviewed-by: Csaba Ringhofer <cs...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
.../apache/impala/service/CatalogOpExecutor.java | 33 +++++++++++++---------
.../java/org/apache/impala/util/DebugUtils.java | 20 +++++++++++++
tests/query_test/test_acid.py | 14 +++++++++
3 files changed, 53 insertions(+), 14 deletions(-)
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 b7726c5e4..811db2cfc 100644
--- a/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
+++ b/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
@@ -6624,24 +6624,29 @@ public class CatalogOpExecutor {
final Timer.Context context
= table.getMetrics().getTimer(HdfsTable.CATALOG_UPDATE_DURATION_METRIC).time();
- long transactionId = -1;
- TblTransaction tblTxn = null;
- if (update.isSetTransaction_id()) {
- transactionId = update.getTransaction_id();
- Preconditions.checkState(transactionId > 0);
- try (MetaStoreClient msClient = catalog_.getMetaStoreClient()) {
- // Setup transactional parameters needed to do alter table/partitions later.
- // TODO: Could be optimized to possibly save some RPCs, as these parameters are
- // not always needed + the writeId of the INSERT could be probably reused.
- tblTxn = MetastoreShim.createTblTransaction(
- msClient.getHiveClient(), table.getMetaStoreTable(), transactionId);
- }
- }
-
try {
// Get new catalog version for table in insert.
long newCatalogVersion = catalog_.incrementAndGetCatalogVersion();
catalog_.getLock().writeLock().unlock();
+
+ TblTransaction tblTxn = null;
+ if (update.isSetTransaction_id()) {
+ long transactionId = update.getTransaction_id();
+ Preconditions.checkState(transactionId > 0);
+ try (MetaStoreClient msClient = catalog_.getMetaStoreClient()) {
+ if (DebugUtils.hasDebugAction(update.getDebug_action(),
+ DebugUtils.UPDATE_CATALOG_ABORT_INSERT_TXN)) {
+ MetastoreShim.abortTransaction(msClient.getHiveClient(), transactionId);
+ LOG.info("Aborted txn due to the debug action.");
+ }
+ // Setup transactional parameters needed to do alter table/partitions later.
+ // TODO: Could be optimized to possibly save some RPCs, as these parameters are
+ // not always needed + the writeId of the INSERT could be probably reused.
+ tblTxn = MetastoreShim.createTblTransaction(
+ msClient.getHiveClient(), table.getMetaStoreTable(), transactionId);
+ }
+ }
+
// Collects the cache directive IDs of any cached table/partitions that were
// targeted. A watch on these cache directives is submitted to the
// TableLoadingMgr and the table will be refreshed asynchronously after all
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 5d2b353da..97cd2b522 100644
--- a/fe/src/main/java/org/apache/impala/util/DebugUtils.java
+++ b/fe/src/main/java/org/apache/impala/util/DebugUtils.java
@@ -54,6 +54,26 @@ public class DebugUtils {
// CatalogOpExecutor#updateCatalog() finishes.
public static final String INSERT_FINISH_DELAY = "catalogd_insert_finish_delay";
+ // debug action label to abort the transaction in updateCatalog.
+ public static final String UPDATE_CATALOG_ABORT_INSERT_TXN =
+ "catalogd_update_catalog_abort_txn";
+
+ /**
+ * Returns true if the label of action is set in the debugActions
+ */
+ public static boolean hasDebugAction(String debugActions, String label) {
+ if (Strings.isNullOrEmpty(debugActions)) {
+ return false;
+ }
+ List<String> actions = Splitter.on('|').splitToList(debugActions);
+ for (String action : actions) {
+ List<String> components = Splitter.on(':').splitToList(action);
+ if (components.isEmpty()) continue;
+ if (components.get(0).equalsIgnoreCase(label)) return true;
+ }
+ return false;
+ }
+
/**
* Given list of debug actions, execute the debug action pertaining to the given label.
* The debugActions string is of the format specified for the query_option/configuration
diff --git a/tests/query_test/test_acid.py b/tests/query_test/test_acid.py
index 53d52270a..98931119f 100644
--- a/tests/query_test/test_acid.py
+++ b/tests/query_test/test_acid.py
@@ -346,3 +346,17 @@ class TestAcid(ImpalaTestSuite):
valid_write_ids = acid_util.get_valid_write_ids(db_name, tbl_name)
new_write_id = valid_write_ids.tblValidWriteIds[0].writeIdHighWaterMark
assert new_write_id > orig_write_id
+
+ @SkipIfHive2.acid
+ def test_alloc_write_id_error_handing(self, unique_database):
+ tbl_name = "insert_only_table"
+ self.client.execute("""CREATE TABLE {0}.{1} (i int)
+ TBLPROPERTIES('transactional'='true', 'transactional_properties'='insert_only')
+ """.format(unique_database, tbl_name))
+ self.execute_query_expect_failure(
+ self.client,
+ "INSERT INTO {0}.{1} VALUES (0), (1), (2)".format(unique_database, tbl_name),
+ {"debug_action": "catalogd_update_catalog_abort_txn"})
+ # Create a new table and load it in catalogd. Catalogd should not hang.
+ self.client.execute("CREATE TABLE {0}.tbl (i int)".format(unique_database))
+ self.client.execute("DESCRIBE {0}.tbl".format(unique_database))