[pull] master from apache:master

.github/workflows/codeql.yml

@@ -37,7 +37,7 @@ jobs:
     steps:
       - name: Checkout repository
         uses: actions/checkout@v7
-      - uses: dorny/paths-filter@fbd0ab8f3e69293af611ebaee6363fc25e6d187d
+      - uses: dorny/paths-filter@ceb8a2b8f2d89434be7ff52d3de7ec3738c5cc9d
         id: filter
         with:
           filters: |

it/mongodb/build.gradle

@@ -36,6 +36,7 @@ dependencies {
     implementation library.java.google_code_gson
     implementation library.java.mongo_java_driver
     implementation library.java.mongo_bson
+    implementation library.java.mongodb_driver_core
     implementation library.java.vendored_guava_32_1_2_jre
 
     testImplementation library.java.mockito_core

it/mongodb/src/main/java/org/apache/beam/it/mongodb/MongoDBResourceManager.java

@@ -20,6 +20,8 @@
 import static org.apache.beam.it.mongodb.MongoDBResourceManagerUtils.checkValidCollectionName;
 import static org.apache.beam.it.mongodb.MongoDBResourceManagerUtils.generateDatabaseName;
 
+import com.mongodb.ConnectionString;
+import com.mongodb.MongoDriverInformation;
 import com.mongodb.client.FindIterable;
 import com.mongodb.client.MongoClient;
 import com.mongodb.client.MongoClients;
@@ -57,6 +59,10 @@ public class MongoDBResourceManager extends TestContainerResourceManager<MongoDB
 
   private static final String DEFAULT_MONGODB_CONTAINER_NAME = "mongo";
 
+  @VisibleForTesting
+  static final MongoDriverInformation DRIVER_INFO =
+      MongoDriverInformation.builder().driverName("Apache Beam").build();
+
   // A list of available MongoDB Docker image tags can be found at
   // https://hub.docker.com/_/mongo/tags
   private static final String DEFAULT_MONGODB_CONTAINER_TAG = "4.0.18";
@@ -88,7 +94,10 @@ private MongoDBResourceManager(Builder builder) {
         usingStaticDatabase ? builder.databaseName : generateDatabaseName(builder.testId);
     this.connectionString =
         String.format("mongodb://%s:%d", this.getHost(), this.getPort(MONGODB_INTERNAL_PORT));
-    this.mongoClient = mongoClient == null ? MongoClients.create(connectionString) : mongoClient;
+    this.mongoClient =
+        mongoClient == null
+            ? MongoClients.create(new ConnectionString(connectionString), DRIVER_INFO)
+            : mongoClient;
   }
 
   public static Builder builder(String testId) {

it/mongodb/src/test/java/org/apache/beam/it/mongodb/MongoDBResourceManagerTest.java

@@ -71,6 +71,11 @@ public void setUp() {
         new MongoDBResourceManager(mongoClient, container, MongoDBResourceManager.builder(TEST_ID));
   }
 
+  @Test
+  public void testDriverInfoHasExpectedName() {
+    assertThat(MongoDBResourceManager.DRIVER_INFO.getDriverNames()).contains("Apache Beam");
+  }
+
   @Test
   public void testCreateResourceManagerBuilderReturnsMongoDBResourceManager() {
     assertThat(

sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/MongoDbGridFSIO.java

@@ -23,6 +23,7 @@
 import com.google.auto.value.AutoValue;
 import com.mongodb.ConnectionString;
 import com.mongodb.MongoClientSettings;
+import com.mongodb.MongoDriverInformation;
 import com.mongodb.client.MongoClient;
 import com.mongodb.client.MongoClients;
 import com.mongodb.client.MongoCursor;
@@ -119,6 +120,9 @@
  */
 public class MongoDbGridFSIO {
 
+  private static final MongoDriverInformation DRIVER_INFO =
+      MongoDriverInformation.builder().driverName("Apache Beam").build();
+
   /** Callback for the parser to use to submit data. */
   public interface ParserCallback<T> extends Serializable {
     /** Output the object. The default timestamp will be the GridFSFile creation timestamp. */
@@ -203,13 +207,13 @@ static ConnectionConfiguration create(
 
     MongoClient setupMongo() {
       if (uri() == null) {
-        return MongoClients.create();
+        return MongoClients.create(MongoClientSettings.builder().build(), DRIVER_INFO);
       }
       MongoClientSettings settings =
           MongoClientSettings.builder()
               .applyConnectionString(new ConnectionString(Preconditions.checkStateNotNull(uri())))
               .build();
-      return MongoClients.create(settings);
+      return MongoClients.create(settings, DRIVER_INFO);
     }
 
     GridFSBucket setupGridFS(MongoClient mongo) {

sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/MongoDbIO.java

@@ -27,6 +27,7 @@
 import com.mongodb.MongoClientSettings;
 import com.mongodb.MongoClientSettings.Builder;
 import com.mongodb.MongoCommandException;
+import com.mongodb.MongoDriverInformation;
 import com.mongodb.client.AggregateIterable;
 import com.mongodb.client.MongoClient;
 import com.mongodb.client.MongoClients;
@@ -145,6 +146,9 @@ public class MongoDbIO {
 
   private static final Logger LOG = LoggerFactory.getLogger(MongoDbIO.class);
 
+  private static final MongoDriverInformation DRIVER_INFO =
+      MongoDriverInformation.builder().driverName("Apache Beam").build();
+
   public static final String ERROR_MSG_QUERY_FN =
       " class is not supported. "
           + "Please provide one of the predefined classes in the MongoDbIO package "
@@ -427,7 +431,7 @@ long getDocumentCount() {
                   spec.ignoreSSLCertificate())
               .applyConnectionString(new ConnectionString(uri))
               .build();
-      try (MongoClient mongoClient = MongoClients.create(settings)) {
+      try (MongoClient mongoClient = MongoClients.create(settings, DRIVER_INFO)) {
         return getDocumentCount(mongoClient, database, collection);
       } catch (Exception e) {
         return -1;
@@ -459,7 +463,7 @@ public long getEstimatedSizeBytes(PipelineOptions pipelineOptions) {
                   spec.ignoreSSLCertificate())
               .applyConnectionString(new ConnectionString(uri))
               .build();
-      try (MongoClient mongoClient = MongoClients.create(settings)) {
+      try (MongoClient mongoClient = MongoClients.create(settings, DRIVER_INFO)) {
         try {
           return getEstimatedSizeBytes(mongoClient, database, collection);
         } catch (MongoCommandException exception) {
@@ -496,7 +500,7 @@ public List<BoundedSource<Document>> split(
                   spec.ignoreSSLCertificate())
               .applyConnectionString(new ConnectionString(uri))
               .build();
-      try (MongoClient mongoClient = MongoClients.create(settings)) {
+      try (MongoClient mongoClient = MongoClients.create(settings, DRIVER_INFO)) {
         MongoDatabase mongoDatabase = mongoClient.getDatabase(database);
 
         List<Document> splitKeys;
@@ -812,7 +816,7 @@ private MongoClient createClient(Read spec) {
                   spec.ignoreSSLCertificate())
               .applyConnectionString(new ConnectionString(uri))
               .build();
-      return MongoClients.create(settings);
+      return MongoClients.create(settings, DRIVER_INFO);
     }
   }
 
@@ -1012,7 +1016,7 @@ public void createMongoClient() {
                     spec.ignoreSSLCertificate())
                 .applyConnectionString(new ConnectionString(uri))
                 .build();
-        client = MongoClients.create(settings);
+        client = MongoClients.create(settings, DRIVER_INFO);
       }
 
       @StartBundle

sdks/java/io/rrio/src/main/java/org/apache/beam/io/requestresponse/Call.java

@@ -596,13 +596,10 @@ private void executeAsync(Callable<Void> callable) throws UserCodeExecutionExcep
   private static <T> void parseAndThrow(Future<T> future, ExecutionException e)
       throws UserCodeExecutionException {
     future.cancel(true);
-    if (e.getCause() == null) {
-      throw new UserCodeExecutionException(e);
+    Throwable cause = e.getCause();
+    if (cause instanceof UserCodeExecutionException) {
+      throw (UserCodeExecutionException) cause;
     }
-    Throwable cause = checkStateNotNull(e.getCause());
-    if (cause instanceof UserCodeQuotaException) {
-      throw new UserCodeQuotaException(cause);
-    }
-    throw new UserCodeExecutionException(cause);
+    throw new UserCodeExecutionException(cause == null ? e : cause);
   }
 }

sdks/java/io/rrio/src/test/java/org/apache/beam/io/requestresponse/CallTest.java

@@ -104,6 +104,20 @@ public void givenCallerThrowsUserCodeExecutionException_emitsIntoFailurePCollect
     pipeline.run();
   }
 
+  @Test
+  public void givenCallerThrowsNonUserCodeException_emitsWrappedUserCodeExecutionException() {
+    Result<Response> result =
+        pipeline
+            .apply(Create.of(new Request("a")))
+            .apply(Call.of(new CallerThrowsRuntimeException(), NON_DETERMINISTIC_RESPONSE_CODER));
+
+    PCollection<ApiIOError> failures = result.getFailures();
+    PAssert.thatSingleton(countStackTracesOf(failures, UserCodeExecutionException.class))
+        .isEqualTo(1L);
+
+    pipeline.run();
+  }
+
   @Test
   public void givenCallerThrowsQuotaException_emitsIntoFailurePCollection() {
     Result<Response> result =
@@ -142,22 +156,39 @@ public void givenCallerTimeout_emitsFailurePCollection() {
   }
 
   @Test
-  public void givenCallerThrowsTimeoutException_emitsFailurePCollection() {
+  public void givenCallerThrowsTimeoutException_thenPreservesExceptionType() {
     Result<Response> result =
         pipeline
             .apply(Create.of(new Request("a")))
             .apply(Call.of(new CallerThrowsTimeout(), NON_DETERMINISTIC_RESPONSE_CODER));
 
     PCollection<ApiIOError> failures = result.getFailures();
     PAssert.thatSingleton(countStackTracesOf(failures, UserCodeExecutionException.class))
-        .isEqualTo(1L);
+        .isEqualTo(0L);
     PAssert.thatSingleton(countStackTracesOf(failures, UserCodeQuotaException.class)).isEqualTo(0L);
     PAssert.thatSingleton(countStackTracesOf(failures, UserCodeTimeoutException.class))
         .isEqualTo(1L);
 
     pipeline.run();
   }
 
+  @Test
+  public void givenCallerThrowsRemoteSystemException_thenPreservesExceptionType() {
+    Result<Response> result =
+        pipeline
+            .apply(Create.of(new Request("a")))
+            .apply(
+                Call.of(new CallerThrowsRemoteSystemException(), NON_DETERMINISTIC_RESPONSE_CODER));
+
+    PCollection<ApiIOError> failures = result.getFailures();
+    PAssert.thatSingleton(countStackTracesOf(failures, UserCodeRemoteSystemException.class))
+        .isEqualTo(1L);
+    PAssert.thatSingleton(countStackTracesOf(failures, UserCodeExecutionException.class))
+        .isEqualTo(0L);
+
+    pipeline.run();
+  }
+
   @Test
   public void givenSetupThrowsUserCodeExecutionException_throwsError() {
     pipeline
@@ -375,6 +406,14 @@ public Response call(Request request) throws UserCodeExecutionException {
     }
   }
 
+  private static class CallerThrowsRuntimeException implements Caller<Request, Response> {
+
+    @Override
+    public Response call(Request request) {
+      throw new RuntimeException("unexpected error");
+    }
+  }
+
   private static class CallerThrowsTimeout implements Caller<Request, Response> {
 
     @Override
@@ -383,6 +422,14 @@ public Response call(Request request) throws UserCodeExecutionException {
     }
   }
 
+  private static class CallerThrowsRemoteSystemException implements Caller<Request, Response> {
+
+    @Override
+    public Response call(Request request) throws UserCodeExecutionException {
+      throw new UserCodeRemoteSystemException("");
+    }
+  }
+
   private static class CallerInvokesQuotaException implements Caller<Request, Response> {
 
     @Override

sdks/python/apache_beam/io/gcp/bigquery_tools.py

@@ -125,6 +125,13 @@
     "GEOGRAPHY": str,
 }
 
+# Duplicated logic with io/gcp/bigquery_change_history.py
+# Default table expiration for auto-created temp datasets: 24 hours in ms.
+# Tables created in the dataset auto-expire after this duration if not
+# explicitly deleted, acting as a safety net for orphaned temp tables
+# (e.g. pipeline crash before cleanup runs).
+_DEFAULT_TABLE_EXPIRATION_MS = 24 * 60 * 60 * 1000
+
 
 class FileFormat(object):
   CSV = 'CSV'
@@ -952,6 +959,7 @@ def create_temporary_dataset(
         project_id,
         self.temp_dataset_id,
         location=location,
+        default_table_expiration_ms=_DEFAULT_TABLE_EXPIRATION_MS,
         labels=labels,
         kms_key=kms_key)