diff --git a/.github/workflows/codeql.yml b/.github/workflows/codeql.yml index 0173d4d7cc4a..98cf5ab88b56 100644 --- a/.github/workflows/codeql.yml +++ b/.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: | diff --git a/it/mongodb/build.gradle b/it/mongodb/build.gradle index 960e15af8394..0a78bdda7724 100644 --- a/it/mongodb/build.gradle +++ b/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 diff --git a/it/mongodb/src/main/java/org/apache/beam/it/mongodb/MongoDBResourceManager.java b/it/mongodb/src/main/java/org/apache/beam/it/mongodb/MongoDBResourceManager.java index 8a4f116c8436..0e11f4b8e293 100644 --- a/it/mongodb/src/main/java/org/apache/beam/it/mongodb/MongoDBResourceManager.java +++ b/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 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) { diff --git a/sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/MongoDbIO.java b/sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/MongoDbIO.java index fc2a761b99a8..46c3f8fcd58a 100644 --- a/sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/MongoDbIO.java +++ b/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> 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 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 diff --git a/sdks/java/io/rrio/src/main/java/org/apache/beam/io/requestresponse/Call.java b/sdks/java/io/rrio/src/main/java/org/apache/beam/io/requestresponse/Call.java index 616a178d1c39..5b9fa4968429 100644 --- a/sdks/java/io/rrio/src/main/java/org/apache/beam/io/requestresponse/Call.java +++ b/sdks/java/io/rrio/src/main/java/org/apache/beam/io/requestresponse/Call.java @@ -596,13 +596,10 @@ private void executeAsync(Callable callable) throws UserCodeExecutionExcep private static void parseAndThrow(Future 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); } } diff --git a/sdks/java/io/rrio/src/test/java/org/apache/beam/io/requestresponse/CallTest.java b/sdks/java/io/rrio/src/test/java/org/apache/beam/io/requestresponse/CallTest.java index 0764ab8db405..5fb20bf38b8b 100644 --- a/sdks/java/io/rrio/src/test/java/org/apache/beam/io/requestresponse/CallTest.java +++ b/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 result = + pipeline + .apply(Create.of(new Request("a"))) + .apply(Call.of(new CallerThrowsRuntimeException(), NON_DETERMINISTIC_RESPONSE_CODER)); + + PCollection failures = result.getFailures(); + PAssert.thatSingleton(countStackTracesOf(failures, UserCodeExecutionException.class)) + .isEqualTo(1L); + + pipeline.run(); + } + @Test public void givenCallerThrowsQuotaException_emitsIntoFailurePCollection() { Result result = @@ -142,7 +156,7 @@ public void givenCallerTimeout_emitsFailurePCollection() { } @Test - public void givenCallerThrowsTimeoutException_emitsFailurePCollection() { + public void givenCallerThrowsTimeoutException_thenPreservesExceptionType() { Result result = pipeline .apply(Create.of(new Request("a"))) @@ -150,7 +164,7 @@ public void givenCallerThrowsTimeoutException_emitsFailurePCollection() { PCollection 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); @@ -158,6 +172,23 @@ public void givenCallerThrowsTimeoutException_emitsFailurePCollection() { pipeline.run(); } + @Test + public void givenCallerThrowsRemoteSystemException_thenPreservesExceptionType() { + Result result = + pipeline + .apply(Create.of(new Request("a"))) + .apply( + Call.of(new CallerThrowsRemoteSystemException(), NON_DETERMINISTIC_RESPONSE_CODER)); + + PCollection 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 { + + @Override + public Response call(Request request) { + throw new RuntimeException("unexpected error"); + } + } + private static class CallerThrowsTimeout implements Caller { @Override @@ -383,6 +422,14 @@ public Response call(Request request) throws UserCodeExecutionException { } } + private static class CallerThrowsRemoteSystemException implements Caller { + + @Override + public Response call(Request request) throws UserCodeExecutionException { + throw new UserCodeRemoteSystemException(""); + } + } + private static class CallerInvokesQuotaException implements Caller { @Override diff --git a/sdks/python/apache_beam/io/gcp/bigquery_tools.py b/sdks/python/apache_beam/io/gcp/bigquery_tools.py index 8dd58cd55a01..0d62ec5233c1 100644 --- a/sdks/python/apache_beam/io/gcp/bigquery_tools.py +++ b/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)