You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by cn...@apache.org on 2026/03/25 15:55:30 UTC
(hadoop) branch branch-3.4 updated: YARN-11952. FederationQueryRunner SQL cleanup
This is an automated email from the ASF dual-hosted git repository.
cnauroth pushed a commit to branch branch-3.4
in repository https://gitbox.apache.org/repos/asf/hadoop.git
The following commit(s) were added to refs/heads/branch-3.4 by this push:
new 4000244fb97 YARN-11952. FederationQueryRunner SQL cleanup
4000244fb97 is described below
commit 4000244fb970c5b1975241ebc9d396e0b112b9f6
Author: Steve Loughran <st...@cloudera.com>
AuthorDate: Wed Mar 25 15:53:44 2026 +0000
YARN-11952. FederationQueryRunner SQL cleanup
All SQL statements other than truncateTable() now use prepared statements.
Extra tests for the public operation which was the most brittle.
Closes #8376
Signed-off-by: Chris Nauroth <cn...@apache.org>
---
.../store/sql/FederationQueryRunner.java | 113 ++++++++++-----------
.../utils/FederationPolicyStoreInputValidator.java | 4 +
.../store/impl/FederationStateStoreBaseTest.java | 40 ++++++++
3 files changed, 100 insertions(+), 57 deletions(-)
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/sql/FederationQueryRunner.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/sql/FederationQueryRunner.java
index 4ff56eef01e..2294ec6a031 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/sql/FederationQueryRunner.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/sql/FederationQueryRunner.java
@@ -21,6 +21,7 @@
import org.apache.hadoop.classification.VisibleForTesting;
import java.sql.Connection;
+import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Statement;
import java.sql.CallableStatement;
@@ -43,15 +44,15 @@ public class FederationQueryRunner {
public final static String YARN_ROUTER_CURRENT_KEY_ID = "YARN_ROUTER_CURRENT_KEY_ID";
public final static String QUERY_SEQUENCE_TABLE_SQL =
- "SELECT nextVal FROM sequenceTable WHERE sequenceName = %s";
+ "SELECT nextVal FROM sequenceTable WHERE sequenceName = ?";
public final static String INSERT_SEQUENCE_TABLE_SQL =
- "INSERT INTO sequenceTable(sequenceName, nextVal) VALUES(%s, %d)";
+ "INSERT INTO sequenceTable(sequenceName, nextVal) VALUES(?, ?)";
public final static String UPDATE_SEQUENCE_TABLE_SQL =
- "UPDATE sequenceTable SET nextVal = %d WHERE sequenceName = %s";
+ "UPDATE sequenceTable SET nextVal = ? WHERE sequenceName = ?";
- public final static String DELETE_QUEUE_SQL = "DELETE FROM policies WHERE queue = %s";
+ public final static String DELETE_QUEUE_SQL = "DELETE FROM policies WHERE queue = ?";
public static final Logger LOG = LoggerFactory.getLogger(FederationQueryRunner.class);
@@ -215,38 +216,40 @@ public int selectOrUpdateSequenceTable(Connection connection, String sequenceNam
int maxSequenceValue = 0;
boolean insertDone = false;
boolean committed = false;
- Statement statement = null;
try {
+ DbType dbType = DatabaseProduct.getDbType(connection);
+ // Build the FOR UPDATE variant of the SELECT template once (string op on the
+ // parameterized constant, before binding so the ? placeholder is preserved).
+ String forUpdateSQL = DatabaseProduct.addForUpdateClause(dbType, QUERY_SEQUENCE_TABLE_SQL);
// Step1. Query SequenceValue.
while (maxSequenceValue == 0) {
- // Query SQL.
- String sql = String.format(QUERY_SEQUENCE_TABLE_SQL, quoteString(sequenceName));
- DbType dbType = DatabaseProduct.getDbType(connection);
- String forUpdateSQL = DatabaseProduct.addForUpdateClause(dbType, sql);
- statement = connection.createStatement();
- ResultSet rs = statement.executeQuery(forUpdateSQL);
- if (rs.next()) {
- maxSequenceValue = rs.getInt("nextVal");
- } else if (insertDone) {
- throw new SQLException("Invalid state of SEQUENCE_TABLE for " + sequenceName);
- } else {
- insertDone = true;
- close(statement);
- statement = connection.createStatement();
- String insertSQL = String.format(INSERT_SEQUENCE_TABLE_SQL, quoteString(sequenceName), 1);
- try {
- statement.executeUpdate(insertSQL);
- } catch (SQLException e) {
- // If the record is already inserted by some other thread continue to select.
- if (isDuplicateKeyError(dbType, e)) {
- continue;
+ try (PreparedStatement select = connection.prepareStatement(forUpdateSQL)) {
+ select.setString(1, sequenceName);
+ try (ResultSet rs = select.executeQuery()) {
+ if (rs.next()) {
+ maxSequenceValue = rs.getInt("nextVal");
+ } else if (insertDone) {
+ throw new SQLException("Invalid state of SEQUENCE_TABLE for " + sequenceName);
+ } else {
+ insertDone = true;
+ try (PreparedStatement insert =
+ connection.prepareStatement(INSERT_SEQUENCE_TABLE_SQL)) {
+ insert.setString(1, sequenceName);
+ insert.setInt(2, 1);
+ try {
+ insert.executeUpdate();
+ } catch (SQLException e) {
+ // If the record is already inserted by some other thread continue to select.
+ if (isDuplicateKeyError(dbType, e)) {
+ continue;
+ }
+ LOG.error("Unable to insert into SEQUENCE_TABLE for {}.", sequenceName, e);
+ throw e;
+ }
+ }
}
- LOG.error("Unable to insert into SEQUENCE_TABLE for {}.", sequenceName, e);
- throw e;
- } finally {
- close(statement);
}
}
}
@@ -254,36 +257,34 @@ public int selectOrUpdateSequenceTable(Connection connection, String sequenceNam
// Step2. Increase SequenceValue.
if (isUpdate) {
int nextSequenceValue = maxSequenceValue + 1;
- close(statement);
- statement = connection.createStatement();
- String updateSQL =
- String.format(UPDATE_SEQUENCE_TABLE_SQL, nextSequenceValue, quoteString(sequenceName));
- statement.executeUpdate(updateSQL);
- maxSequenceValue = nextSequenceValue;
+ try (PreparedStatement update =
+ connection.prepareStatement(UPDATE_SEQUENCE_TABLE_SQL)) {
+ update.setInt(1, nextSequenceValue);
+ update.setString(2, sequenceName);
+ update.executeUpdate();
+ maxSequenceValue = nextSequenceValue;
+ }
}
connection.commit();
committed = true;
return maxSequenceValue;
- } catch(SQLException e){
+ } catch (SQLException e) {
throw new SQLException("Unable to selectOrUpdateSequenceTable due to: " + e.getMessage(), e);
} finally {
if (!committed) {
rollbackDBConn(connection);
}
- close(statement);
}
}
public void updateSequenceTable(Connection connection, String sequenceName, int sequenceValue)
throws SQLException {
- String updateSQL =
- String.format(UPDATE_SEQUENCE_TABLE_SQL, sequenceValue, quoteString(sequenceName));
boolean committed = false;
- Statement statement = null;
- try {
- statement = connection.createStatement();
- statement.executeUpdate(updateSQL);
+ try (PreparedStatement statement = connection.prepareStatement(UPDATE_SEQUENCE_TABLE_SQL)) {
+ statement.setInt(1, sequenceValue);
+ statement.setString(2, sequenceName);
+ statement.executeUpdate();
connection.commit();
committed = true;
} catch (SQLException e) {
@@ -292,18 +293,21 @@ public void updateSequenceTable(Connection connection, String sequenceName, int
if (!committed) {
rollbackDBConn(connection);
}
- close(statement);
}
}
+ /**
+ * Drop a queue from the policy.
+ * @param connection DB connection
+ * @param queue queue name
+ * @throws SQLException failure
+ */
public void deletePolicyByQueue(Connection connection, String queue)
throws SQLException {
- String deleteSQL = String.format(DELETE_QUEUE_SQL, quoteString(queue));
boolean committed = false;
- Statement statement = null;
- try {
- statement = connection.createStatement();
- statement.executeUpdate(deleteSQL);
+ try (PreparedStatement statement = connection.prepareStatement(DELETE_QUEUE_SQL)) {
+ statement.setString(1, queue);
+ statement.executeUpdate();
connection.commit();
committed = true;
} catch (SQLException e) {
@@ -312,7 +316,6 @@ public void deletePolicyByQueue(Connection connection, String queue)
if (!committed) {
rollbackDBConn(connection);
}
- close(statement);
}
}
@@ -339,9 +342,9 @@ public void truncateTable(Connection connection, String tableName)
private String getTruncateStatement(DbType dbType, String tableName) {
if (isMYSQL(dbType)) {
- return ("DELETE FROM \"" + tableName + "\"");
+ return "DELETE FROM `" + tableName + "`";
} else {
- return("DELETE FROM " + tableName);
+ return "DELETE FROM " + tableName;
}
}
@@ -358,8 +361,4 @@ static void rollbackDBConn(Connection dbConn) {
LOG.warn("Failed to rollback db connection ", e);
}
}
-
- static String quoteString(String input) {
- return "'" + input + "'";
- }
}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/utils/FederationPolicyStoreInputValidator.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/utils/FederationPolicyStoreInputValidator.java
index e3d7ee03c1b..3d3081e8bc6 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/utils/FederationPolicyStoreInputValidator.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/utils/FederationPolicyStoreInputValidator.java
@@ -158,5 +158,9 @@ public static void validate(DeleteSubClusterPoliciesConfigurationsRequest reques
throw new FederationStateStoreInvalidInputException(
"The queues that needs to be deleted cannot be empty.");
}
+
+ for (String queue : queues) {
+ checkQueue(queue);
+ }
}
}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/store/impl/FederationStateStoreBaseTest.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/store/impl/FederationStateStoreBaseTest.java
index a0073e0575c..9b7147be11c 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/store/impl/FederationStateStoreBaseTest.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/store/impl/FederationStateStoreBaseTest.java
@@ -26,11 +26,13 @@
import java.util.Set;
import java.util.HashSet;
import java.util.TimeZone;
+import java.util.stream.Collectors;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.security.token.delegation.DelegationKey;
import org.apache.hadoop.test.LambdaTestUtils;
+import org.apache.hadoop.util.Lists;
import org.apache.hadoop.util.Time;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.ReservationId;
@@ -89,6 +91,8 @@
import org.apache.hadoop.yarn.server.federation.store.records.RouterRMTokenResponse;
import org.apache.hadoop.yarn.server.records.Version;
import org.apache.hadoop.yarn.util.MonotonicClock;
+
+import org.assertj.core.api.Assertions;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
@@ -1189,6 +1193,42 @@ public void testDeletePoliciesConfigurations() throws Exception {
}
}
+ @Test
+ public void testAccidentalQueueDeletion() throws Exception {
+ // Make sure the parameters are processed.
+ FederationStateStore federationStateStore = this.getStateStore();
+ setPolicyConf("Queue1", "PolicyType1");
+ setPolicyConf("Queue2", "PolicyType2");
+
+ // with a PreparedStatement it is treated as a literal queue name that does not match.
+ List<String> queues = Lists.newArrayList("' OR '1'='1");
+ federationStateStore.deletePoliciesConfigurations(
+ DeleteSubClusterPoliciesConfigurationsRequest.newInstance(queues));
+
+ // Both queues must still be present.
+ GetSubClusterPoliciesConfigurationsResponse response =
+ federationStateStore.getPoliciesConfigurations(
+ GetSubClusterPoliciesConfigurationsRequest.newInstance());
+ Assertions.assertThat(response).describedAs("response").isNotNull();
+ List<String> remaining = response.getPoliciesConfigs()
+ .stream()
+ .map(SubClusterPolicyConfiguration::getQueue)
+ .collect(Collectors.toList());
+ Assertions.assertThat(remaining)
+ .contains("Queue1");
+ Assertions.assertThat(remaining)
+ .contains("Queue2");
+ }
+
+ @Test
+ public void testEmptyQueueNameRejected() throws Exception {
+ // The input validator must reject a request that contains a blank queue name.
+ DeleteSubClusterPoliciesConfigurationsRequest request =
+ DeleteSubClusterPoliciesConfigurationsRequest.newInstance(Lists.newArrayList("ValidQueue", ""));
+ LambdaTestUtils.intercept(YarnException.class, "Missing Queue",
+ () -> getStateStore().deletePoliciesConfigurations(request));
+ }
+
@Test
public void testDeletePolicyStore() throws Exception {
// Step1. We delete all Policies Configurations.
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org