Fix test

.github/trigger_files/beam_PostCommit_Python.json

@@ -1,6 +1,6 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run.",
   "https://github.com/apache/beam/pull/32440": "test new datastream runner for batch"
-  "modification": 8
+  "modification": 9
 }
 

.github/trigger_files/beam_PostCommit_Python_Xlang_Gcp_Dataflow.json

@@ -1,4 +1,4 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run",
-  "modification": 11
+  "modification": 10
 }

.github/trigger_files/beam_PostCommit_Python_Xlang_Gcp_Direct.json

@@ -1,4 +1,4 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run",
-  "modification": 4
+  "modification": 5
 }

sdks/python/apache_beam/io/external/xlang_jdbcio_it_test.py

@@ -19,11 +19,13 @@
 
 import datetime
 import logging
+import os
 import time
 import typing
 import unittest
 from decimal import Decimal
 
+import pytest
 from parameterized import parameterized
 
 import apache_beam as beam
@@ -88,13 +90,24 @@
 coders.registry.register_coder(SimpleRow, coders.RowCoder)
 
 
+@pytest.mark.uses_gcp_java_expansion_service
+@unittest.skipUnless(
+    os.environ.get('EXPANSION_JARS'),
+    "EXPANSION_JARS environment var is not provided, "
+    "indicating that jars have not been built")
 @unittest.skipIf(sqlalchemy is None, 'sql alchemy package is not installed.')
 @unittest.skipIf(
     PostgresContainer is None, 'testcontainers package is not installed')
 @unittest.skipIf(
     TestPipeline().get_pipeline_options().view_as(StandardOptions).runner is
     None,
     'Do not run this test on precommit suites.')
+@unittest.skipIf(
+    TestPipeline().get_pipeline_options().view_as(StandardOptions).runner is
+    not None and
+    "dataflowrunner" in TestPipeline().get_pipeline_options().view_as(
+        StandardOptions).runner.lower(),
+    'Do not run this test on dataflow runner.')
 class CrossLanguageJdbcIOTest(unittest.TestCase):
   DbData = typing.NamedTuple(
       'DbData',
@@ -436,8 +449,7 @@ def test_xlang_jdbc_custom_statements(self, database):
               partition_column="id",
               schema=SimpleRow))
 
-      # revert
-      assert_that(result, equal_to(test_rows))
+      assert_that(result, equal_to(expected_filtered_rows))
 
 
 if __name__ == '__main__':

sdks/python/test-suites/portable/common.gradle

@@ -385,18 +385,19 @@ project.tasks.register("postCommitPy${pythonVersionSuffix}IT") {
   doLast {
     def tests = [
             "apache_beam/io/gcp/bigquery_read_it_test.py",
-            "apache_beam/io/external/xlang_jdbcio_it_test.py",
             "apache_beam/io/external/xlang_kafkaio_it_test.py",
             "apache_beam/io/external/xlang_kinesisio_it_test.py",
             "apache_beam/io/external/xlang_debeziumio_it_test.py",
     ]
     def testOpts = ["${tests.join(' ')}"] + ["--log-cli-level=INFO"]
+    def flink_conf_dir = "${rootDir}/runners/flink/src/test/resources/"
     def pipelineOpts = [
         "--runner=FlinkRunner",
         "--project=apache-beam-testing",
         "--environment_type=LOOPBACK",
         "--temp_location=gs://temp-storage-for-end-to-end-tests/temp-it",
         "--flink_job_server_jar=${project(":runners:flink:${latestFlinkVersion}:job-server").shadowJar.archivePath}",
+        "--flink_conf_dir=${flink_conf_dir}",
         '--sdk_harness_log_level_overrides=' +
             // suppress info level flink.runtime log flood
             '{\\"org.apache.flink.runtime\\":\\"WARN\\",' +