DatasetDao: map JDBC driver NPE on versionSchema column read to NotFound

ambry-account/src/main/java/com/github/ambry/account/mysql/DatasetDao.java

@@ -1850,7 +1850,24 @@ private Dataset executeGetDatasetStatement(PreparedStatement statement, int acco
             "Dataset expired for account: " + accountId + " container: " + containerId + " dataset: " + datasetName,
             AccountServiceErrorCode.Deleted);
       }
-      versionSchema = Dataset.VersionSchema.values()[resultSet.getInt(VERSION_SCHEMA)];
+      int versionSchemaOrdinal;
+      try {
+        versionSchemaOrdinal = resultSet.getInt(VERSION_SCHEMA);
+      } catch (NullPointerException e) {
+        // mysql-connector-java 8.0.21 (see gradle/dependency-versions.gradle) can throw NPE
+        // from ResultSetImpl.findColumn -> getInt when the row state is unexpected. Observed
+        // in prod 2026-04-29 on /named/<account>/<container>/<dataset>, where
+        // it surfaced as HTTP 500 instead of 404. No fix is called out in the 8.0.x release
+        // notes; revisit when bumping the connector. This path is MySQL-only -- Ambry's
+        // dataset metadata does not run against TiDB. Map to NotFound so the frontend
+        // translates it to 404; the dedicated metric below is the on-call signal (the
+        // existing AccountAndContainerInjector error log already names the dataset).
+        metrics.datasetRowReadNpeCount.inc();
+        throw new AccountServiceException(
+            "Dataset row could not be read for account: " + accountId + " container: " + containerId + " dataset: "
+                + datasetName, AccountServiceErrorCode.NotFound);
+      }
+      versionSchema = Dataset.VersionSchema.values()[versionSchemaOrdinal];
       retentionPolicy = resultSet.getObject(RETENTION_POLICY, String.class);
       retentionCount = resultSet.getObject(RETENTION_COUNT, Integer.class);
       retentionTimeInSeconds = resultSet.getObject(RETENTION_TIME_IN_SECONDS, Long.class);
@@ -1914,7 +1931,18 @@ private Dataset.VersionSchema executeGetVersionSchema(PreparedStatement statemen
             "Version Schema not found for account: " + accountId + " container: " + containerId + " dataset: "
                 + datasetName, AccountServiceErrorCode.NotFound);
       }
-      versionSchema = Dataset.VersionSchema.values()[resultSet.getInt(VERSION_SCHEMA)];
+      int versionSchemaOrdinal;
+      try {
+        versionSchemaOrdinal = resultSet.getInt(VERSION_SCHEMA);
+      } catch (NullPointerException e) {
+        // Same JDBC driver bug as executeGetDatasetStatement -- see comment there for details.
+        // Not observed on this path in prod, but the call shape is identical so guard symmetrically.
+        metrics.datasetRowReadNpeCount.inc();
+        throw new AccountServiceException(
+            "Version Schema row could not be read for account: " + accountId + " container: " + containerId
+                + " dataset: " + datasetName, AccountServiceErrorCode.NotFound);
+      }
+      versionSchema = Dataset.VersionSchema.values()[versionSchemaOrdinal];
     } finally {
       //If result set is not created in a try-with-resources block, it needs to be closed in a finally block.
       closeQuietly(resultSet);

ambry-account/src/test/java/com/github/ambry/account/mysql/DatasetDaoTest.java

@@ -0,0 +1,130 @@
+/*
+ * Copyright 2026 LinkedIn Corp. All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ */
+package com.github.ambry.account.mysql;
+
+import com.codahale.metrics.MetricRegistry;
+import com.github.ambry.account.AccountServiceErrorCode;
+import com.github.ambry.account.AccountServiceException;
+import com.github.ambry.config.MySqlAccountServiceConfig;
+import com.github.ambry.config.VerifiableProperties;
+import com.github.ambry.mysql.MySqlDataAccessor;
+import com.github.ambry.mysql.MySqlMetrics;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.Timestamp;
+import java.util.Properties;
+import org.junit.Test;
+
+import static com.github.ambry.account.mysql.AccountDaoTest.getDataAccessor;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.ArgumentMatchers.contains;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+
+/** Unit test for {@link DatasetDao}. */
+public class DatasetDaoTest {
+
+  /**
+   * Regression for prod incident on 2026-04-29: a NPE thrown from the MySQL JDBC driver
+   * while reading the {@code versionSchema} column (ResultSetImpl.findColumn -> getInt)
+   * surfaced as HTTP 500 from {@code GET /named/<account>/<container>/<dataset>}
+   * because the exception bubbled past {@link DatasetDao#getDataset} uncaught. Verify the
+   * narrow catch translates it into {@link AccountServiceException} with
+   * {@link AccountServiceErrorCode#NotFound}, increments the operability counter, and
+   * keeps the dataset identifiers in the exception message.
+   */
+  @Test
+  public void testGetDatasetMapsJdbcNpeOnVersionSchemaToNotFound() throws Exception {
+    Connection mockConnection = mock(Connection.class);
+    MySqlMetrics metrics = new MySqlMetrics(DatasetDao.class, new MetricRegistry());
+    MySqlDataAccessor dataAccessor = getDataAccessor(mockConnection, metrics);
+    PreparedStatement mockGetDatasetStatement = mock(PreparedStatement.class);
+    when(mockConnection.prepareStatement(contains("from " + DatasetDao.DATASET_TABLE))).thenReturn(
+        mockGetDatasetStatement);
+
+    ResultSet mockResultSet = mock(ResultSet.class);
+    when(mockResultSet.next()).thenReturn(true);
+    // Future-dated deletion ts so the deleted-check passes and we reach the column read.
+    when(mockResultSet.getTimestamp(eq(DatasetDao.DELETED_TS))).thenReturn(
+        new Timestamp(System.currentTimeMillis() + 60_000L));
+    // Reproduce the JDBC driver NPE seen in prod.
+    when(mockResultSet.getInt(eq(DatasetDao.VERSION_SCHEMA))).thenThrow(new NullPointerException());
+    when(mockGetDatasetStatement.executeQuery()).thenReturn(mockResultSet);
+
+    Properties props = new Properties();
+    props.setProperty(MySqlAccountServiceConfig.DB_INFO, "");
+    MySqlAccountServiceConfig config = new MySqlAccountServiceConfig(new VerifiableProperties(props));
+    DatasetDao dao = new DatasetDao(dataAccessor, config, metrics);
+
+    long npeCountBefore = metrics.datasetRowReadNpeCount.getCount();
+    try {
+      dao.getDataset(1, 2, "acct", "cont", "ds");
+      fail("Expected AccountServiceException");
+    } catch (AccountServiceException e) {
+      assertEquals(AccountServiceErrorCode.NotFound, e.getErrorCode());
+      String message = e.getMessage();
+      assertTrue("message should include accountId, got: " + message, message.contains("1"));
+      assertTrue("message should include containerId, got: " + message, message.contains("2"));
+      assertTrue("message should include datasetName, got: " + message, message.contains("ds"));
+    }
+    assertEquals("datasetRowReadNpeCount should increment by 1", npeCountBefore + 1,
+        metrics.datasetRowReadNpeCount.getCount());
+  }
+
+  /**
+   * Same JDBC NPE class as {@link #testGetDatasetMapsJdbcNpeOnVersionSchemaToNotFound}, but
+   * exercising the {@code executeGetVersionSchema} path used by version-mutation flows
+   * (delete/list/ttl-update). Not observed in prod, but the call shape is identical so the
+   * mapping should be symmetric.
+   */
+  @Test
+  public void testDeleteDatasetVersionMapsJdbcNpeOnVersionSchemaToNotFound() throws Exception {
+    Connection mockConnection = mock(Connection.class);
+    MySqlMetrics metrics = new MySqlMetrics(DatasetDao.class, new MetricRegistry());
+    MySqlDataAccessor dataAccessor = getDataAccessor(mockConnection, metrics);
+    PreparedStatement mockGetVersionSchemaStatement = mock(PreparedStatement.class);
+    // getVersionSchemaSql is "select versionSchema from Datasets where ..." -- match on the table.
+    when(mockConnection.prepareStatement(contains("from " + DatasetDao.DATASET_TABLE))).thenReturn(
+        mockGetVersionSchemaStatement);
+
+    ResultSet mockResultSet = mock(ResultSet.class);
+    when(mockResultSet.next()).thenReturn(true);
+    when(mockResultSet.getInt(eq(DatasetDao.VERSION_SCHEMA))).thenThrow(new NullPointerException());
+    when(mockGetVersionSchemaStatement.executeQuery()).thenReturn(mockResultSet);
+
+    Properties props = new Properties();
+    props.setProperty(MySqlAccountServiceConfig.DB_INFO, "");
+    MySqlAccountServiceConfig config = new MySqlAccountServiceConfig(new VerifiableProperties(props));
+    DatasetDao dao = new DatasetDao(dataAccessor, config, metrics);
+
+    long npeCountBefore = metrics.datasetRowReadNpeCount.getCount();
+    try {
+      dao.deleteDatasetVersion(1, 2, "ds", "v1");
+      fail("Expected AccountServiceException");
+    } catch (AccountServiceException e) {
+      assertEquals(AccountServiceErrorCode.NotFound, e.getErrorCode());
+      String message = e.getMessage();
+      assertTrue("message should include accountId, got: " + message, message.contains("account: 1"));
+      assertTrue("message should include containerId, got: " + message, message.contains("container: 2"));
+      assertTrue("message should include datasetName, got: " + message, message.contains("dataset: ds"));
+    }
+    assertEquals("datasetRowReadNpeCount should increment by 1", npeCountBefore + 1,
+        metrics.datasetRowReadNpeCount.getCount());
+  }
+}

ambry-mysql/src/main/java/com/github/ambry/mysql/MySqlMetrics.java

@@ -42,6 +42,7 @@ public class MySqlMetrics {
   public static final String CONNECTION_SUCCESS_COUNT = "ConnectionSuccessCount";
   public static final String CONNECTION_FAILURE_COUNT = "ConnectionFailureCount";
   public static final String CONSTRUCT_RETENTION_POLICY_FAILURE_COUNT = "ConstructRetentionPolicyFailureCount";
+  public static final String DATASET_ROW_READ_NPE_COUNT = "DatasetRowReadNpeCount";
 
   public final Histogram writeTimeMs;
   public final Counter writeSuccessCount;
@@ -67,6 +68,7 @@ public class MySqlMetrics {
   public final Counter connectionFailureCount;
 
   public final Counter constructRetentionPolicyFailureCount;
+  public final Counter datasetRowReadNpeCount;
 
   public MySqlMetrics(Class<?> clazz, MetricRegistry metricRegistry) {
     writeTimeMs = metricRegistry.histogram(MetricRegistry.name(clazz, WRITE_TIME_MSEC));
@@ -88,5 +90,6 @@ public MySqlMetrics(Class<?> clazz, MetricRegistry metricRegistry) {
     connectionFailureCount = metricRegistry.counter(MetricRegistry.name(clazz, CONNECTION_FAILURE_COUNT));
     constructRetentionPolicyFailureCount =
         metricRegistry.counter(MetricRegistry.name(clazz, CONSTRUCT_RETENTION_POLICY_FAILURE_COUNT));
+    datasetRowReadNpeCount = metricRegistry.counter(MetricRegistry.name(clazz, DATASET_ROW_READ_NPE_COUNT));
   }
 }