From d23303a5e4f2b50fabbaca7b4c1ea13d4b8c6d1a Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Sun, 28 Jun 2026 18:01:13 +0000 Subject: [PATCH 1/7] try to fix iceberg post commit test --- .../extended_tests/databases/iceberg.yaml | 54 ++++++++------ .../apache_beam/yaml/integration_tests.py | 70 +++++++++++++++++++ 2 files changed, 102 insertions(+), 22 deletions(-) diff --git a/sdks/python/apache_beam/yaml/extended_tests/databases/iceberg.yaml b/sdks/python/apache_beam/yaml/extended_tests/databases/iceberg.yaml index d7449233aab5..98462f2d0271 100644 --- a/sdks/python/apache_beam/yaml/extended_tests/databases/iceberg.yaml +++ b/sdks/python/apache_beam/yaml/extended_tests/databases/iceberg.yaml @@ -16,8 +16,18 @@ # fixtures: - - name: TEMP_DIR - type: "tempfile.TemporaryDirectory" + - name: ICEBERG_FIXTURE + type: "apache_beam.yaml.integration_tests.temp_iceberg_table_with_pk" + config: + table_data: + name: "labels" + schema: + type: "struct" + schema-id: 0 + fields: + - { id: 1, name: "label", required: true, type: "string" } + - { id: 2, name: "rank", required: true, type: "long" } + identifier-field-ids: [1] pipelines: - name: write @@ -32,14 +42,14 @@ pipelines: - {label: "389a", rank: 2} - type: WriteToIceberg config: - table: db.labels - catalog_name: hadoop_catalog + table: "{ICEBERG_FIXTURE[table]}" + catalog_name: rest_catalog catalog_properties: - type: hadoop - warehouse: "{TEMP_DIR}" + type: rest + uri: "{ICEBERG_FIXTURE[api_url]}" options: project: "apache-beam-testing" - temp_location: "{TEMP_DIR}" + temp_location: "{ICEBERG_FIXTURE[temp_dir]}" - name: read pipeline: @@ -47,11 +57,11 @@ pipelines: transforms: - type: ReadFromIceberg config: - table: db.labels - catalog_name: hadoop_catalog + table: "{ICEBERG_FIXTURE[table]}" + catalog_name: rest_catalog catalog_properties: - type: hadoop - warehouse: "{TEMP_DIR}" + type: rest + uri: "{ICEBERG_FIXTURE[api_url]}" - type: AssertEqual config: elements: @@ -60,7 +70,7 @@ pipelines: - {label: "389a", rank: 2} options: project: "apache-beam-testing" - temp_location: "{TEMP_DIR}" + temp_location: "{ICEBERG_FIXTURE[temp_dir]}" - name: read_cdc_batch pipeline: @@ -68,11 +78,11 @@ pipelines: transforms: - type: ReadFromIcebergCDC config: - table: db.labels - catalog_name: hadoop_catalog + table: "{ICEBERG_FIXTURE[table]}" + catalog_name: rest_catalog catalog_properties: - type: hadoop - warehouse: "{TEMP_DIR}" + type: rest + uri: "{ICEBERG_FIXTURE[api_url]}" from_timestamp: 1762819200000 to_timestamp: 2078352000000 filter: '"label" = ''11a'' or "rank" = 1' @@ -86,7 +96,7 @@ pipelines: - {label: "37a", rank: 1} options: project: "apache-beam-testing" - temp_location: "{TEMP_DIR}" + temp_location: "{ICEBERG_FIXTURE[temp_dir]}" - name: read_cdc_streaming pipeline: @@ -94,11 +104,11 @@ pipelines: transforms: - type: ReadFromIcebergCDC config: - table: db.labels - catalog_name: hadoop_catalog + table: "{ICEBERG_FIXTURE[table]}" + catalog_name: rest_catalog catalog_properties: - type: hadoop - warehouse: "{TEMP_DIR}" + type: rest + uri: "{ICEBERG_FIXTURE[api_url]}" streaming: True to_timestamp: 2078352000000 filter: '"label" = ''11a'' or "rank" = 1' @@ -112,4 +122,4 @@ pipelines: - {label: "37a", rank: 1} options: project: "apache-beam-testing" - temp_location: "{TEMP_DIR}" + temp_location: "{ICEBERG_FIXTURE[temp_dir]}" diff --git a/sdks/python/apache_beam/yaml/integration_tests.py b/sdks/python/apache_beam/yaml/integration_tests.py index e319a3d3a9bd..5bcd2f91d522 100644 --- a/sdks/python/apache_beam/yaml/integration_tests.py +++ b/sdks/python/apache_beam/yaml/integration_tests.py @@ -25,10 +25,14 @@ import logging import os import random +import requests import secrets +import shutil import sqlite3 import string import struct +import tempfile +import time import unittest import uuid from datetime import datetime @@ -39,6 +43,7 @@ from apache_beam.coders import Coder from apache_beam.coders.coder_impl import CoderImpl from apache_beam.yaml.test_utils.datadog_test_utils import temp_fake_datadog_server +from testcontainers.core.container import DockerContainer class BigEndianIntegerCoderImpl(CoderImpl): @@ -556,6 +561,71 @@ def temp_oracle_database(): yield f"jdbc:oracle:thin:system/oracle@localhost:{port}/XEPDB1" +@contextlib.contextmanager +def temp_iceberg_table_with_pk(table_data): + + # Create a temp dir that will be shared between host and container. + # We use the exact same path on both to avoid path mapping issues. + temp_dir = tempfile.mkdtemp() + os.chmod(temp_dir, 0o777) + + # Start the Iceberg REST catalog container + container = DockerContainer("tabulario/iceberg-rest:0.6.0") + container.with_exposed_ports(8181) + container.with_volume_mapping(temp_dir, temp_dir, mode='rw') + container.with_env("HADOOP_USER_NAME", "iceberg") + container.with_env("CATALOG_WAREHOUSE", temp_dir) + container.with_env( + "CATALOG_IO__IMPL", "org.apache.iceberg.hadoop.HadoopFileIO") + + try: + container.start() + + ip = container.get_container_host_ip() + port = container.get_exposed_port(8181) + api_url = f"http://{ip}:{port}" + + # Poll the REST API until it is ready + for _ in range(30): + try: + response = requests.get(f"{api_url}/v1/config") + if response.status_code == 200: + break + except requests.exceptions.ConnectionError: + pass + time.sleep(1) + else: + raise RuntimeError("Iceberg REST catalog failed to start in time.") + + # Create namespace 'db' + requests.post( + f"{api_url}/v1/namespaces", + json={"namespace": ["db"]}, + headers={"Content-Type": "application/json"}) + + # Create table with primary key + response = requests.post( + f"{api_url}/v1/namespaces/db/tables", + json=table_data, + headers={"Content-Type": "application/json"}) + if response.status_code != 200: + raise RuntimeError(f"Failed to create Iceberg table: {response.text}") + + # Change permissions of the created directories inside the container + # so the host user can write to them. + container.get_wrapped_container().exec_run(f"chmod -R 777 {temp_dir}") + + yield { + "api_url": api_url, + "temp_dir": temp_dir, + "table": f"db.{table_data['name']}" + } + + finally: + container.stop() + shutil.rmtree(temp_dir, ignore_errors=True) + + @contextlib.contextmanager def temp_kafka_server(): """Context manager to provide a temporary Kafka server for testing. From 98330233d8b9117d758dc30c2e87298c56c44d9b Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Sun, 28 Jun 2026 18:01:47 +0000 Subject: [PATCH 2/7] trigger file --- .github/trigger_files/beam_PostCommit_Yaml_Xlang_Direct.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/trigger_files/beam_PostCommit_Yaml_Xlang_Direct.json b/.github/trigger_files/beam_PostCommit_Yaml_Xlang_Direct.json index b5704c67ef1c..86bf1193abd9 100644 --- a/.github/trigger_files/beam_PostCommit_Yaml_Xlang_Direct.json +++ b/.github/trigger_files/beam_PostCommit_Yaml_Xlang_Direct.json @@ -1,4 +1,4 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run", - "revision": 6 + "revision": 7 } From 46ac56a391e0ed99fea912f6b45347ee3e9bf497 Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Sun, 28 Jun 2026 18:41:04 +0000 Subject: [PATCH 3/7] isort --- .../apache_beam/yaml/integration_tests.py | 24 +++++++++---------- 1 file changed, 11 insertions(+), 13 deletions(-) diff --git a/sdks/python/apache_beam/yaml/integration_tests.py b/sdks/python/apache_beam/yaml/integration_tests.py index 5bcd2f91d522..c81cc4b1a919 100644 --- a/sdks/python/apache_beam/yaml/integration_tests.py +++ b/sdks/python/apache_beam/yaml/integration_tests.py @@ -25,7 +25,6 @@ import logging import os import random -import requests import secrets import shutil import sqlite3 @@ -39,7 +38,7 @@ from datetime import timezone import mock - +import requests from apache_beam.coders import Coder from apache_beam.coders.coder_impl import CoderImpl from apache_beam.yaml.test_utils.datadog_test_utils import temp_fake_datadog_server @@ -72,10 +71,20 @@ def get_impl(self): 'beam:coders:javasdk:0.1', None, lambda payload, components, context: BigEndianIntegerCoder()) +import apache_beam as beam import psycopg2 import pytds import sqlalchemy import yaml +from apache_beam.io import filesystems +from apache_beam.io.gcp.bigquery_tools import BigQueryWrapper +from apache_beam.io.gcp.internal.clients import bigquery +from apache_beam.io.gcp.spanner_wrapper import SpannerWrapper +from apache_beam.options.pipeline_options import PipelineOptions +from apache_beam.utils import python_callable +from apache_beam.yaml import yaml_provider +from apache_beam.yaml import yaml_transform +from apache_beam.yaml.conftest import yaml_test_files_dir from apitools.base.py.exceptions import HttpError from google.cloud import pubsub_v1 from google.cloud.bigtable import client @@ -89,17 +98,6 @@ def get_impl(self): from testcontainers.mysql import MySqlContainer from testcontainers.postgres import PostgresContainer -import apache_beam as beam -from apache_beam.io import filesystems -from apache_beam.io.gcp.bigquery_tools import BigQueryWrapper -from apache_beam.io.gcp.internal.clients import bigquery -from apache_beam.io.gcp.spanner_wrapper import SpannerWrapper -from apache_beam.options.pipeline_options import PipelineOptions -from apache_beam.utils import python_callable -from apache_beam.yaml import yaml_provider -from apache_beam.yaml import yaml_transform -from apache_beam.yaml.conftest import yaml_test_files_dir - _LOGGER = logging.getLogger(__name__) _MONGO_CONTAINER_IMAGE = 'mongo:7.0.7' From 9f5255993363e77796f0c9abd7f9ef8f1bd557ab Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Sun, 28 Jun 2026 19:18:22 +0000 Subject: [PATCH 4/7] oops more lint --- .../apache_beam/yaml/integration_tests.py | 25 +++++++++++-------- 1 file changed, 14 insertions(+), 11 deletions(-) diff --git a/sdks/python/apache_beam/yaml/integration_tests.py b/sdks/python/apache_beam/yaml/integration_tests.py index c81cc4b1a919..c1b084f9a659 100644 --- a/sdks/python/apache_beam/yaml/integration_tests.py +++ b/sdks/python/apache_beam/yaml/integration_tests.py @@ -39,10 +39,11 @@ import mock import requests +from testcontainers.core.container import DockerContainer + from apache_beam.coders import Coder from apache_beam.coders.coder_impl import CoderImpl from apache_beam.yaml.test_utils.datadog_test_utils import temp_fake_datadog_server -from testcontainers.core.container import DockerContainer class BigEndianIntegerCoderImpl(CoderImpl): @@ -71,20 +72,11 @@ def get_impl(self): 'beam:coders:javasdk:0.1', None, lambda payload, components, context: BigEndianIntegerCoder()) -import apache_beam as beam import psycopg2 import pytds import sqlalchemy import yaml -from apache_beam.io import filesystems -from apache_beam.io.gcp.bigquery_tools import BigQueryWrapper -from apache_beam.io.gcp.internal.clients import bigquery -from apache_beam.io.gcp.spanner_wrapper import SpannerWrapper -from apache_beam.options.pipeline_options import PipelineOptions -from apache_beam.utils import python_callable -from apache_beam.yaml import yaml_provider -from apache_beam.yaml import yaml_transform -from apache_beam.yaml.conftest import yaml_test_files_dir + from apitools.base.py.exceptions import HttpError from google.cloud import pubsub_v1 from google.cloud.bigtable import client @@ -98,6 +90,17 @@ def get_impl(self): from testcontainers.mysql import MySqlContainer from testcontainers.postgres import PostgresContainer +import apache_beam as beam +from apache_beam.io import filesystems +from apache_beam.io.gcp.bigquery_tools import BigQueryWrapper +from apache_beam.io.gcp.internal.clients import bigquery +from apache_beam.io.gcp.spanner_wrapper import SpannerWrapper +from apache_beam.options.pipeline_options import PipelineOptions +from apache_beam.utils import python_callable +from apache_beam.yaml import yaml_provider +from apache_beam.yaml import yaml_transform +from apache_beam.yaml.conftest import yaml_test_files_dir + _LOGGER = logging.getLogger(__name__) _MONGO_CONTAINER_IMAGE = 'mongo:7.0.7' From 8502bea453ab66543eac82ca6db654b652eff52b Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Sun, 28 Jun 2026 19:48:03 +0000 Subject: [PATCH 5/7] lint --- sdks/python/apache_beam/yaml/integration_tests.py | 1 - 1 file changed, 1 deletion(-) diff --git a/sdks/python/apache_beam/yaml/integration_tests.py b/sdks/python/apache_beam/yaml/integration_tests.py index c1b084f9a659..86759149fe77 100644 --- a/sdks/python/apache_beam/yaml/integration_tests.py +++ b/sdks/python/apache_beam/yaml/integration_tests.py @@ -76,7 +76,6 @@ def get_impl(self): import pytds import sqlalchemy import yaml - from apitools.base.py.exceptions import HttpError from google.cloud import pubsub_v1 from google.cloud.bigtable import client From 57d72db5d6d50c9c942b0b18662d85336d828f14 Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Sun, 28 Jun 2026 21:30:17 +0000 Subject: [PATCH 6/7] fix tmp dir --- sdks/python/apache_beam/yaml/integration_tests.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/sdks/python/apache_beam/yaml/integration_tests.py b/sdks/python/apache_beam/yaml/integration_tests.py index 86759149fe77..2d0135ebaa99 100644 --- a/sdks/python/apache_beam/yaml/integration_tests.py +++ b/sdks/python/apache_beam/yaml/integration_tests.py @@ -566,7 +566,9 @@ def temp_iceberg_table_with_pk(table_data): # Create a temp dir that will be shared between host and container. # We use the exact same path on both to avoid path mapping issues. - temp_dir = tempfile.mkdtemp() + # We create it in the current working directory (workspace) because + # Docker in GitHub Actions often cannot mount directories from /tmp. + temp_dir = tempfile.mkdtemp(dir=os.getcwd()) os.chmod(temp_dir, 0o777) # Start the Iceberg REST catalog container From 93c2ab85724aba4bc76ecc8fb115b919182724ad Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Sun, 28 Jun 2026 23:22:40 +0000 Subject: [PATCH 7/7] add timeouts per gemini comments --- sdks/python/apache_beam/yaml/integration_tests.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/sdks/python/apache_beam/yaml/integration_tests.py b/sdks/python/apache_beam/yaml/integration_tests.py index 2d0135ebaa99..a7abd223b295 100644 --- a/sdks/python/apache_beam/yaml/integration_tests.py +++ b/sdks/python/apache_beam/yaml/integration_tests.py @@ -590,10 +590,10 @@ def temp_iceberg_table_with_pk(table_data): # Poll the REST API until it is ready for _ in range(30): try: - response = requests.get(f"{api_url}/v1/config") + response = requests.get(f"{api_url}/v1/config", timeout=5) if response.status_code == 200: break - except requests.exceptions.ConnectionError: + except (requests.exceptions.ConnectionError, requests.exceptions.Timeout): pass time.sleep(1) else: @@ -603,13 +603,15 @@ def temp_iceberg_table_with_pk(table_data): requests.post( f"{api_url}/v1/namespaces", json={"namespace": ["db"]}, - headers={"Content-Type": "application/json"}) + headers={"Content-Type": "application/json"}, + timeout=10) # Create table with primary key response = requests.post( f"{api_url}/v1/namespaces/db/tables", json=table_data, - headers={"Content-Type": "application/json"}) + headers={"Content-Type": "application/json"}, + timeout=10) if response.status_code != 200: raise RuntimeError(f"Failed to create Iceberg table: {response.text}")