Fix triggeringFrequency documentation and validation for STORAGE_API_AT_LEAST_ONCE

- Updated JavaDoc to reflect that triggeringFrequency applies to FILE_LOADS,
  STORAGE_WRITE_API, and STORAGE_API_AT_LEAST_ONCE methods
- Fixed validation logic in expand() to require triggeringFrequency for
  STORAGE_API_AT_LEAST_ONCE when writing unbounded PCollections
- Removed conflicting warning that incorrectly stated STORAGE_API_AT_LEAST_ONCE
  ignores triggeringFrequency
- Added comprehensive test cases to verify proper validation behavior:
  * Test failure when triggeringFrequency is missing for unbounded collections
  * Test success when triggeringFrequency is provided for unbounded collections
  * Test success for bounded collections without triggeringFrequency requirement
- Updated error messages to include all three supported methods

This ensures consistent behavior across all BigQuery write methods that support
triggered writes for unbounded collections.

Author: Arunodoy18

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java

@@ -3330,8 +3330,9 @@ public Write<T> withLoadJobProjectId(ValueProvider<String> loadJobProjectId) {
     /**
      * Choose the frequency at which file writes are triggered.
      *
-     * <p>This is only applicable when the write method is set to {@link Method#FILE_LOADS} or
-     * {@link Method#STORAGE_WRITE_API}, and only when writing an unbounded {@link PCollection}.
+     * <p>This is only applicable when the write method is set to {@link Method#FILE_LOADS}, {@link
+     * Method#STORAGE_WRITE_API}, or {@link Method#STORAGE_API_AT_LEAST_ONCE}, and only when writing
+     * an unbounded {@link PCollection}.
      *
      * <p>Every triggeringFrequency duration, a BigQuery load job will be generated for all the data
      * written since the last load job. BigQuery has limits on how many load jobs can be triggered
@@ -3736,19 +3737,22 @@ public WriteResult expand(PCollection<T> input) {
       BigQueryOptions bqOptions = input.getPipeline().getOptions().as(BigQueryOptions.class);
       Write.Method method = resolveMethod(input);
       if (input.isBounded() == IsBounded.UNBOUNDED) {
-        if (method == Write.Method.FILE_LOADS || method == Write.Method.STORAGE_WRITE_API) {
+        if (method == Write.Method.FILE_LOADS
+            || method == Write.Method.STORAGE_WRITE_API
+            || method == Write.Method.STORAGE_API_AT_LEAST_ONCE) {
           Duration triggeringFrequency =
-              (method == Write.Method.STORAGE_WRITE_API)
+              (method == Write.Method.STORAGE_WRITE_API
+                      || method == Write.Method.STORAGE_API_AT_LEAST_ONCE)
                   ? getStorageApiTriggeringFrequency(bqOptions)
                   : getTriggeringFrequency();
           checkArgument(
               triggeringFrequency != null,
-              "When writing an unbounded PCollection via FILE_LOADS or STORAGE_WRITE_API, "
+              "When writing an unbounded PCollection via FILE_LOADS, STORAGE_WRITE_API, or STORAGE_API_AT_LEAST_ONCE, "
                   + "triggering frequency must be specified");
         } else {
           checkArgument(
               getTriggeringFrequency() == null,
-              "Triggering frequency can be specified only when writing via FILE_LOADS or STORAGE_WRITE_API, but the method was %s.",
+              "Triggering frequency can be specified only when writing via FILE_LOADS, STORAGE_WRITE_API, or STORAGE_API_AT_LEAST_ONCE, but the method was %s.",
               method);
         }
         if (method != Method.FILE_LOADS) {
@@ -3757,13 +3761,7 @@ public WriteResult expand(PCollection<T> input) {
               "Number of file shards can be specified only when writing via FILE_LOADS, but the method was %s.",
               method);
         }
-        if (method == Method.STORAGE_API_AT_LEAST_ONCE
-            && getStorageApiTriggeringFrequency(bqOptions) != null) {
-          LOG.warn(
-              "Storage API triggering frequency option will be ignored is it can only be specified only "
-                  + "when writing via STORAGE_WRITE_API, but the method was {}.",
-              method);
-        }
+
         if (getAutoSharding()) {
           if (method == Method.STORAGE_WRITE_API && getStorageApiNumStreams(bqOptions) > 0) {
             LOG.warn(

sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOWriteTest.java

@@ -2632,7 +2632,8 @@ public void testStreamingWriteValidateFailsWithoutTriggeringFrequency() {
     Method method = useStorageApi ? Method.STORAGE_WRITE_API : Method.FILE_LOADS;
 
     thrown.expect(IllegalArgumentException.class);
-    thrown.expectMessage("unbounded PCollection via FILE_LOADS or STORAGE_WRITE_API");
+    thrown.expectMessage(
+        "unbounded PCollection via FILE_LOADS, STORAGE_WRITE_API, or STORAGE_API_AT_LEAST_ONCE");
     thrown.expectMessage("triggering frequency must be specified");
 
     p.getOptions().as(BigQueryOptions.class).setStorageWriteApiTriggeringFrequencySec(null);
@@ -2646,6 +2647,72 @@ public void testStreamingWriteValidateFailsWithoutTriggeringFrequency() {
                 .withCreateDisposition(CreateDisposition.CREATE_NEVER));
   }
 
+  @Test
+  public void testStreamingWriteValidateFailsWithoutTriggeringFrequencyForStorageApiAtLeastOnce() {
+    assumeTrue(useStreaming);
+    assumeTrue(useStorageApiApproximate); // Test STORAGE_API_AT_LEAST_ONCE specifically
+    p.enableAbandonedNodeEnforcement(false);
+
+    thrown.expect(IllegalArgumentException.class);
+    thrown.expectMessage(
+        "unbounded PCollection via FILE_LOADS, STORAGE_WRITE_API, or STORAGE_API_AT_LEAST_ONCE");
+    thrown.expectMessage("triggering frequency must be specified");
+
+    p.getOptions().as(BigQueryOptions.class).setStorageWriteApiTriggeringFrequencySec(null);
+    p.apply(Create.empty(INPUT_RECORD_CODER))
+        .setIsBoundedInternal(PCollection.IsBounded.UNBOUNDED)
+        .apply(
+            BigQueryIO.<InputRecord>write()
+                .withAvroFormatFunction(r -> new GenericData.Record(r.getSchema()))
+                .to("dataset.table")
+                .withMethod(Method.STORAGE_API_AT_LEAST_ONCE)
+                .withCreateDisposition(CreateDisposition.CREATE_NEVER));
+  }
+
+  @Test
+  public void testStreamingWriteValidateSucceedsWithTriggeringFrequencyForStorageApiAtLeastOnce() {
+    assumeTrue(useStreaming);
+    assumeTrue(useStorageApiApproximate); // Test STORAGE_API_AT_LEAST_ONCE specifically
+    p.enableAbandonedNodeEnforcement(false);
+
+    // This should not throw - STORAGE_API_AT_LEAST_ONCE with triggering frequency should be valid
+    p.getOptions().as(BigQueryOptions.class).setStorageWriteApiTriggeringFrequencySec(30);
+    p.apply(Create.empty(INPUT_RECORD_CODER))
+        .setIsBoundedInternal(PCollection.IsBounded.UNBOUNDED)
+        .apply(
+            BigQueryIO.<InputRecord>write()
+                .withAvroFormatFunction(r -> new GenericData.Record(r.getSchema()))
+                .to("dataset.table")
+                .withMethod(Method.STORAGE_API_AT_LEAST_ONCE)
+                .withCreateDisposition(CreateDisposition.CREATE_NEVER)
+                .withTestServices(fakeBqServices)
+                .withoutValidation());
+    // Should validate without throwing
+    p.run();
+  }
+
+  @Test
+  public void testBoundedWriteValidateSucceedsWithoutTriggeringFrequencyForStorageApiAtLeastOnce() {
+    assumeTrue(!useStreaming); // Test bounded PCollection
+    assumeTrue(useStorageApiApproximate); // Test STORAGE_API_AT_LEAST_ONCE specifically
+
+    // Bounded collections should not require triggering frequency even for
+    // STORAGE_API_AT_LEAST_ONCE
+    p.getOptions().as(BigQueryOptions.class).setStorageWriteApiTriggeringFrequencySec(null);
+    p.apply(Create.empty(INPUT_RECORD_CODER))
+        .setIsBoundedInternal(PCollection.IsBounded.BOUNDED)
+        .apply(
+            BigQueryIO.<InputRecord>write()
+                .withAvroFormatFunction(r -> new GenericData.Record(r.getSchema()))
+                .to("dataset.table")
+                .withMethod(Method.STORAGE_API_AT_LEAST_ONCE)
+                .withCreateDisposition(CreateDisposition.CREATE_NEVER)
+                .withTestServices(fakeBqServices)
+                .withoutValidation());
+    // Should validate without throwing
+    p.run();
+  }
+
   @Test
   public void testBigQueryIOGetName() {
     assertEquals(