diff --git a/.github/trigger_files/beam_PostCommit_Python_Xlang_Messaging_Direct.json b/.github/trigger_files/beam_PostCommit_Python_Xlang_Messaging_Direct.json index 455144f02a35..d6a91b7e2e86 100644 --- a/.github/trigger_files/beam_PostCommit_Python_Xlang_Messaging_Direct.json +++ b/.github/trigger_files/beam_PostCommit_Python_Xlang_Messaging_Direct.json @@ -1,4 +1,4 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run", - "modification": 6 + "modification": 7 } diff --git a/CHANGES.md b/CHANGES.md index b98c46cb8813..fcb011d1489f 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -67,6 +67,7 @@ * Upgraded Iceberg dependency to 1.11.0 (Java) ([#38925](https://github.com/apache/beam/issues/38925)). * Support for X source added (Java/Python) ([#X](https://github.com/apache/beam/issues/X)). * Add ArrowFlight IO (Java) ([#20116](https://github.com/apache/beam/issues/20116)). +* (Python) JmsIO (IBM MQ, ActiveMQ, and other providers) is now supported in Python via cross-language ([#30716](https://github.com/apache/beam/issues/30716)). ## New Features / Improvements @@ -115,6 +116,7 @@ ## Bugfixes +* Fixed unresolved runtime `ValueProvider` options being stringified in Python Dataflow Flex Templates ([#39499](https://github.com/apache/beam/issues/39499)). * Fixed unbounded checkpoint state growth for splittable DoFns that self-checkpoint on the portable Flink runner (Java) ([#27648](https://github.com/apache/beam/issues/27648)). * Improved Java pipeline performance by avoiding repeated `DoFn` type descriptor resolution when creating cached invokers ([#39309](https://github.com/apache/beam/issues/39309)). * (Python) Fixed a memory leak in Python SDK caused by storing exceptions with potentially large stack frames in a cache ([#39406](https://github.com/apache/beam/issues/39406)). diff --git a/contributor-docs/README.md b/contributor-docs/README.md index 1087a74f05c7..7e4381893c93 100644 --- a/contributor-docs/README.md +++ b/contributor-docs/README.md @@ -22,6 +22,7 @@ This directory contains documentation for contributors to the Apache Beam projec - [Committer Guide](committer-guide.md): Guidelines for Beam committers regarding code review, pull request objectives, merging processes, and post-merge tasks. - [Committer Onboarding](committer-onboarding.md): A checklist for new Beam committers to set up their accounts and permissions. - [Java Dependency Upgrades](java-dependency-upgrades.md): Instructions for upgrading Java dependencies in Beam, including running linkage checkers and verification tests. +- [Local Flink Python Validation](local-flink-python.md): Instructions for running Python pipelines on a local Flink standalone cluster. - [Python Tips](python-tips.md): Tips and instructions for developing the Python SDK, including environment setup, running tests, and handling dependencies. - [RC Testing Guide](rc-testing-guide.md): A guide for testing Beam Release Candidates (RCs) against downstream projects for Python, Java, and Go SDKs. - [Release Guide](release-guide.md): A comprehensive guide for the Release Manager on how to perform a Beam release, from preparation to promotion. diff --git a/contributor-docs/local-flink-python.md b/contributor-docs/local-flink-python.md new file mode 100644 index 000000000000..6ab30435c68b --- /dev/null +++ b/contributor-docs/local-flink-python.md @@ -0,0 +1,204 @@ + + +# Running Python pipelines on a local Flink cluster + +This guide describes a contributor workflow for validating Python Beam pipelines +against a real local Flink standalone cluster. It is useful when embedded Flink +is not enough, for example when validating streaming source behavior, checkpoint +boundaries, or runner-visible job state in the Flink dashboard. + +The commands assume a Unix shell (Linux, macOS, or WSL2 on Windows) with `curl`, +`tar`, and `java` on the `PATH`. + +* [What this setup validates](#what-this-setup-validates) +* [Prerequisites](#prerequisites) +* [Start a local Flink cluster](#start-a-local-flink-cluster) +* [Run a Beam Python pipeline](#run-a-beam-python-pipeline) +* [Troubleshooting](#troubleshooting) +* [Stop the cluster](#stop-the-cluster) + +## What this setup validates + +This setup runs three components: + +1. A Flink standalone cluster, consisting of a JobManager and a TaskManager. +1. A Beam Flink Job Server, started by the Python `FlinkRunner`. +1. A Python SDK harness, using `--environment_type=LOOPBACK` for local + development. + +The Flink dashboard at `http://localhost:8081` shows the submitted Beam jobs. +This is different from embedded Flink mode, where the cluster is started only +for the lifetime of one job and is not useful for manual dashboard inspection. + +## Prerequisites + +Install or prepare the following: + +* Docker Desktop (optional), only for the alternative method of obtaining the + Flink distribution. +* A Unix shell: Linux, macOS, or WSL2 on Windows. +* Java 11 on the `PATH`. +* A Python environment with the Beam SDK dependencies installed. +* A Beam source checkout for the Python code under test. +* A Flink 1.20 Job Server jar built from the same Beam checkout when validating + unreleased Beam changes. + +For a source-built Job Server jar, run this command from the Beam checkout: + +```sh +./gradlew :runners:flink:1.20:job-server:shadowJar +``` + +The jar is written under: + +```text +runners/flink/1.20/job-server/build/libs/ +``` + +## Start a local Flink cluster + +Use a Flink distribution whose minor version matches a Flink version supported +by your Beam version. See the [Flink Version Compatibility](https://beam.apache.org/documentation/runners/flink/#flink-version-compatibility) +table in the Flink Runner documentation, and confirm the exact patch version on +the [Flink downloads page](https://flink.apache.org/downloads.html). This guide +uses Flink 1.20. + +Download and unpack the binary distribution: + +```sh +FLINK_VERSION=1.20.1 +curl -fLO "https://archive.apache.org/dist/flink/flink-${FLINK_VERSION}/flink-${FLINK_VERSION}-bin-scala_2.12.tgz" +tar -xzf "flink-${FLINK_VERSION}-bin-scala_2.12.tgz" -C "$HOME" +export FLINK_HOME="$HOME/flink-${FLINK_VERSION}" +``` + +Ensure these settings exist in `$FLINK_HOME/conf/config.yaml`: + +```yaml +jobmanager.rpc.address: localhost +rest.address: localhost +taskmanager.numberOfTaskSlots: 2 +``` + +Start the cluster. The JobManager and TaskManager run as background daemons: + +```sh +"$FLINK_HOME/bin/start-cluster.sh" +``` + +Verify that the JobManager and TaskManager are available: + +```sh +curl -fsS http://localhost:8081/overview +``` + +Expected output includes one TaskManager and two slots: + +```json +{"taskmanagers":1,"slots-total":2,"slots-available":2,"jobs-running":0} +``` + +You can also open the Flink dashboard in a browser: + +```text +http://localhost:8081 +``` + +### Alternative: extract Flink from the Docker image + +If a direct download is not available, copy the distribution out of the Flink +Docker image with `docker cp`: + +```sh +docker create --name flink-dist flink:1.20 +docker cp flink-dist:/opt/flink "$HOME/flink-1.20" +docker rm flink-dist +export FLINK_HOME="$HOME/flink-1.20" +``` + +A distribution copied out of a Docker image can contain the container hostname in +`conf/config.yaml`; see [Troubleshooting](#troubleshooting). + +## Run a Beam Python pipeline + +For local Python development, use `FlinkRunner`, point it at the standalone +cluster, and use `LOOPBACK` so the Python SDK harness runs in the local process. + +Use a source checkout on `PYTHONPATH` when validating unreleased Python changes. +Set paths for your environment: + +```sh +export BEAM_CHECKOUT="$HOME/beam" +export PYTHON="$HOME/beamenv/bin/python" +export FLINK_JOB_SERVER_JAR="$(find "$BEAM_CHECKOUT/runners/flink/1.20/job-server/build/libs" \ + -name 'beam-runners-flink-1.20-job-server-*.jar' | head -n 1)" +``` + +Run a small pipeline: + +```sh +printf 'to be or not to be\nbeam runs on flink\n' > /tmp/beam-flink-input.txt + +PYTHONPATH="$BEAM_CHECKOUT/sdks/python" "$PYTHON" -m apache_beam.examples.wordcount \ + --runner=FlinkRunner \ + --flink_master=localhost:8081 \ + --flink_version=1.20 \ + --flink_job_server_jar="$FLINK_JOB_SERVER_JAR" \ + --environment_type=LOOPBACK \ + --input=/tmp/beam-flink-input.txt \ + --output=/tmp/beam-flink-counts +``` + +For released Beam, omit `--flink_job_server_jar` and the `PYTHONPATH` prefix; the +`FlinkRunner` downloads a Job Server matching `--flink_version` automatically. The +source checkout and built jar are only needed to test unreleased changes. + +Check the dashboard or REST API after the run: + +```sh +curl -fsS http://localhost:8081/jobs/overview +``` + +The job should be `FINISHED`. + +## Troubleshooting + +If the TaskManager does not register, check `$FLINK_HOME/conf/config.yaml`. +When a distribution is copied out of a Docker image, the file might contain the +container hostname. Replace it with: + +```yaml +jobmanager.rpc.address: localhost +``` + +If a Python job fails on native Windows with an invalid path containing `:`, +run the Python driver and Job Server from WSL2. Some staged artifact names used +by the portable runner are valid on Linux but invalid as native Windows file +names. + +On WSL2, keep at least one shell open in the distribution while the cluster runs. +Closing the last shell can stop the distribution and its background daemons. + +If the job starts but the Python transforms do not execute, check the +environment type. `LOOPBACK` is intended for local development. For a remote +or multi-machine Flink cluster, use a containerized environment instead. + +## Stop the cluster + +Stop the local cluster when you finish collecting results: + +```sh +"$FLINK_HOME/bin/stop-cluster.sh" +``` diff --git a/sdks/go/pkg/beam/runners/prism/internal/coders.go b/sdks/go/pkg/beam/runners/prism/internal/coders.go index d326a332b8d3..0f770849a984 100644 --- a/sdks/go/pkg/beam/runners/prism/internal/coders.go +++ b/sdks/go/pkg/beam/runners/prism/internal/coders.go @@ -367,6 +367,17 @@ func pullDecoderNoAlloc(c *pipepb.Coder, coders map[string]*pipepb.Coder) func(i ed(r) wd(r) } + case urns.CoderShardedKey: + ccids := c.GetComponentCoderIds() + if len(ccids) != 1 { + panic(fmt.Sprintf("ShardedKey coder must have only 1 component: %s", prototext.Format(c))) + } + kd := pullDecoderNoAlloc(coders[ccids[0]], coders) + return func(r io.Reader) { + l, _ := coder.DecodeVarInt(r) + ioutilx.ReadN(r, int(l)) + kd(r) + } case urns.CoderRow: panic(fmt.Sprintf("Runner forgot to LP this Row Coder. %v", prototext.Format(c))) default: diff --git a/sdks/go/pkg/beam/runners/prism/internal/coders_test.go b/sdks/go/pkg/beam/runners/prism/internal/coders_test.go index 4656a94e03ec..1d1a8b6d4596 100644 --- a/sdks/go/pkg/beam/runners/prism/internal/coders_test.go +++ b/sdks/go/pkg/beam/runners/prism/internal/coders_test.go @@ -370,6 +370,22 @@ func Test_pullDecoder(t *testing.T) { }, }, []byte{3, 0}, + }, { + "sharded_key", + &pipepb.Coder{ + Spec: &pipepb.FunctionSpec{ + Urn: urns.CoderShardedKey, + }, + ComponentCoderIds: []string{"key"}, + }, + map[string]*pipepb.Coder{ + "key": { + Spec: &pipepb.FunctionSpec{ + Urn: urns.CoderVarInt, + }, + }, + }, + []byte{3, 1, 2, 3, 255, 3}, }, } for _, test := range tests { diff --git a/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsIOTest.java b/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsIOTest.java index eb6fb4faec04..d23b33873e14 100644 --- a/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsIOTest.java +++ b/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsIOTest.java @@ -32,27 +32,20 @@ import static org.hamcrest.Matchers.greaterThanOrEqualTo; import static org.hamcrest.Matchers.hasProperty; import static org.hamcrest.Matchers.is; -import static org.hamcrest.Matchers.isA; import static org.hamcrest.Matchers.lessThan; import static org.hamcrest.core.StringContains.containsString; import static org.hamcrest.object.HasToString.hasToString; import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.anyLong; -import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.verifyNoMoreInteractions; import static org.mockito.Mockito.when; import java.io.IOException; -import java.io.NotSerializableException; import java.io.Serializable; import java.lang.reflect.Proxy; import java.nio.ByteBuffer; @@ -66,8 +59,6 @@ import java.util.HashSet; import java.util.List; import java.util.Set; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Function; import javax.jms.BytesMessage; @@ -84,7 +75,6 @@ import org.apache.activemq.command.ActiveMQMessage; import org.apache.activemq.util.Callback; import org.apache.beam.sdk.PipelineResult; -import org.apache.beam.sdk.coders.Coder; import org.apache.beam.sdk.coders.SerializableCoder; import org.apache.beam.sdk.coders.StringUtf8Coder; import org.apache.beam.sdk.io.UnboundedSource; @@ -94,16 +84,12 @@ import org.apache.beam.sdk.metrics.MetricNameFilter; import org.apache.beam.sdk.metrics.MetricQueryResults; import org.apache.beam.sdk.metrics.MetricsFilter; -import org.apache.beam.sdk.options.ExecutorOptions; -import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.options.PipelineOptionsFactory; -import org.apache.beam.sdk.testing.CoderProperties; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; import org.apache.beam.sdk.transforms.Count; import org.apache.beam.sdk.transforms.Create; import org.apache.beam.sdk.transforms.SerializableBiFunction; -import org.apache.beam.sdk.util.SerializableUtils; import org.apache.beam.sdk.values.PCollection; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Throwables; import org.apache.qpid.jms.JmsAcknowledgeCallback; @@ -117,7 +103,6 @@ import org.junit.Test; import org.junit.runner.RunWith; import org.junit.runners.Parameterized; -import org.mockito.ArgumentCaptor; import org.mockito.Mockito; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -163,9 +148,9 @@ public static Collection connectionFactories() { private ConnectionFactory connectionFactory; private final Class connectionFactoryClass; private ConnectionFactory connectionFactoryWithSyncAcksAndWithoutPrefetch; - private final String brokerUrl; - private final Integer brokerPort; - private final String forceAsyncAcksParam; + private String brokerUrl; + private String forceAsyncAcksParam; + private int brokerPort; public JmsIOTest( String brokerUrl, @@ -252,20 +237,6 @@ public void testReadMessages() throws Exception { assertQueueIsEmpty(); } - @Test - public void testPipelineWithNonSerializableCF() { - SerializableUtils.ensureSerializable( - JmsIO.read() - .withConnectionFactoryProviderFn(__ -> new MockNonSerializableConnectionFactory())); - try { - SerializableUtils.ensureSerializable( - JmsIO.read().withConnectionFactory(new MockNonSerializableConnectionFactory())); - fail(); - } catch (Exception e) { - assertThat(Throwables.getRootCause(e), isA(NotSerializableException.class)); - } - } - @Test public void testReadMessagesWithCFProviderFn() throws Exception { long count = 5; @@ -522,32 +493,6 @@ public void testWriteDynamicMessage() throws Exception { assertEquals(100, count); } - @Test - public void testSplitForQueue() throws Exception { - JmsIO.Read read = JmsIO.read().withQueue(QUEUE); - PipelineOptions pipelineOptions = PipelineOptionsFactory.create(); - int desiredNumSplits = 5; - JmsIO.UnboundedJmsSource initialSource = new JmsIO.UnboundedJmsSource(read); - List splits = initialSource.split(desiredNumSplits, pipelineOptions); - // in the case of a queue, we have concurrent consumers by default, so the initial number - // splits is equal to the desired number of splits - assertEquals(desiredNumSplits, splits.size()); - } - - @Test - public void testSplitForTopic() throws Exception { - JmsIO.Read read = JmsIO.read().withTopic(TOPIC); - PipelineOptions pipelineOptions = PipelineOptionsFactory.create(); - int desiredNumSplits = 5; - JmsIO.UnboundedJmsSource initialSource = new JmsIO.UnboundedJmsSource(read); - List splits = initialSource.split(desiredNumSplits, pipelineOptions); - // in the case of a topic, we can have only a unique subscriber on the topic per pipeline - // else it means we can have duplicate messages (all subscribers on the topic receive every - // message). - // So, whatever the desizedNumSplits is, the actual number of splits should be 1. - assertEquals(1, splits.size()); - } - private boolean advanceWithRetry(UnboundedSource.UnboundedReader reader) throws IOException { for (int attempt = 0; attempt < 10; attempt++) { if (reader.advance()) { @@ -685,63 +630,6 @@ public void testCheckpointMarkAndFinalizeSeparatelyClientAcknowledgeUnsafe() thr assertEquals(5, count(QUEUE)); } - @Test - public void testJmsCheckpointMarkIndividualAcknowledgeAllMessages() throws Exception { - Message msg1 = Mockito.mock(Message.class); - Message msg2 = Mockito.mock(Message.class); - Message msg3 = Mockito.mock(Message.class); - - JmsCheckpointMark.Preparer preparer = - JmsCheckpointMark.newPreparer(JmsIO.AcknowledgeMode.INDIVIDUAL_ACKNOWLEDGE); - preparer.add(msg1); - preparer.add(msg2); - preparer.add(msg3); - - AtomicInteger activeCheckpoints = new AtomicInteger(0); - JmsCheckpointMark mark = - preparer.newCheckpoint( - null, null, JmsIO.AcknowledgeMode.INDIVIDUAL_ACKNOWLEDGE, activeCheckpoints); - assertNotNull(mark.getMessages()); - assertEquals(3, mark.getMessages().size()); - assertNull(mark.getConsumer()); - assertNull(mark.getSession()); - assertEquals(1, activeCheckpoints.get()); - - mark.finalizeCheckpoint(); - - Mockito.verify(msg1, Mockito.times(1)).acknowledge(); - Mockito.verify(msg2, Mockito.times(1)).acknowledge(); - Mockito.verify(msg3, Mockito.times(1)).acknowledge(); - assertEquals(0, activeCheckpoints.get()); - } - - @Test - public void testJmsCheckpointMarkClientAcknowledgeUnsafeNoSessionRecreation() throws Exception { - Message msg1 = Mockito.mock(Message.class); - Message msg2 = Mockito.mock(Message.class); - - JmsCheckpointMark.Preparer preparer = - JmsCheckpointMark.newPreparer(JmsIO.AcknowledgeMode.CLIENT_ACKNOWLEDGE_UNSAFE); - preparer.add(msg1); - preparer.add(msg2); - - AtomicInteger activeCheckpoints = new AtomicInteger(0); - JmsCheckpointMark mark = - preparer.newCheckpoint( - null, null, JmsIO.AcknowledgeMode.CLIENT_ACKNOWLEDGE_UNSAFE, activeCheckpoints); - assertNotNull(mark.getMessages()); - assertEquals(1, mark.getMessages().size()); - assertNull(mark.getConsumer()); - assertNull(mark.getSession()); - assertEquals(1, activeCheckpoints.get()); - - mark.finalizeCheckpoint(); - - Mockito.verify(msg2, Mockito.times(1)).acknowledge(); - Mockito.verify(msg1, Mockito.never()).acknowledge(); - assertEquals(0, activeCheckpoints.get()); - } - private JmsIO.UnboundedJmsReader setupReaderForTest() throws JMSException { return setupReaderForTest(null); } @@ -890,17 +778,6 @@ public void testCheckpointMarkSafety() throws Exception { runner.join(); } - /** Test the checkpoint mark default coder, which is actually AvroCoder. */ - @Test - public void testCheckpointMarkDefaultCoder() throws Exception { - JmsCheckpointMark jmsCheckpointMark = - JmsCheckpointMark.newPreparer(JmsIO.AcknowledgeMode.CLIENT_ACKNOWLEDGE) - .newCheckpoint(null, null, JmsIO.AcknowledgeMode.CLIENT_ACKNOWLEDGE, null); - Coder coder = new JmsIO.UnboundedJmsSource(null).getCheckpointMarkCoder(); - CoderProperties.coderSerializable(coder); - CoderProperties.coderDecodeEncodeEqual(coder, jmsCheckpointMark); - } - @Test public void testDefaultAutoscaler() throws IOException { JmsIO.Read spec = @@ -945,53 +822,6 @@ public void testCustomAutoscaler() throws IOException { verify(autoScaler, times(1)).stop(); } - @Test - public void testCloseWithTimeout() throws IOException, JMSException { - Duration closeTimeout = Duration.millis(2000L); - JmsIO.Read spec = - JmsIO.read() - .withConnectionFactory(connectionFactory) - .withUsername(USERNAME) - .withPassword(PASSWORD) - .withQueue(QUEUE) - .withCloseTimeout(closeTimeout); - - JmsIO.UnboundedJmsSource source = new JmsIO.UnboundedJmsSource(spec); - - ScheduledExecutorService mockScheduledExecutorService = - Mockito.mock(ScheduledExecutorService.class); - ExecutorOptions options = PipelineOptionsFactory.as(ExecutorOptions.class); - options.setScheduledExecutorService(mockScheduledExecutorService); - ArgumentCaptor runnableArgumentCaptor = ArgumentCaptor.forClass(Runnable.class); - when(mockScheduledExecutorService.schedule( - runnableArgumentCaptor.capture(), anyLong(), any(TimeUnit.class))) - .thenReturn(null /* unused */); - - JmsIO.UnboundedJmsReader reader = source.createReader(options, null); - reader.start(); - assertFalse(getDiscardedValue(reader)); - reader.checkpointMarkPreparer.add(Mockito.mock(Message.class)); - CheckpointMark mark = reader.getCheckpointMark(); - reader.close(); - assertTrue(getDiscardedValue(reader)); - verify(mockScheduledExecutorService) - .schedule(any(Runnable.class), eq(1L), eq(TimeUnit.SECONDS)); - mark.finalizeCheckpoint(); - runnableArgumentCaptor.getValue().run(); - assertTrue(getDiscardedValue(reader)); - verifyNoMoreInteractions(mockScheduledExecutorService); - } - - private boolean getDiscardedValue(JmsIO.UnboundedJmsReader reader) { - JmsCheckpointMark.Preparer preparer = reader.checkpointMarkPreparer; - preparer.lock.readLock().lock(); - try { - return preparer.discarded; - } finally { - preparer.lock.readLock().unlock(); - } - } - @Test public void testDiscardCheckpointMark() throws Exception { Connection connection = diff --git a/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsLocalTest.java b/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsLocalTest.java new file mode 100644 index 000000000000..4ea8f6d317a7 --- /dev/null +++ b/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsLocalTest.java @@ -0,0 +1,245 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.jms; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.isA; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.io.IOException; +import java.io.NotSerializableException; +import java.util.List; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import javax.jms.Connection; +import javax.jms.ConnectionFactory; +import javax.jms.JMSException; +import javax.jms.Message; +import javax.jms.MessageConsumer; +import javax.jms.Queue; +import javax.jms.Session; +import org.apache.activemq.ActiveMQConnectionFactory; +import org.apache.beam.sdk.coders.Coder; +import org.apache.beam.sdk.options.ExecutorOptions; +import org.apache.beam.sdk.options.PipelineOptions; +import org.apache.beam.sdk.options.PipelineOptionsFactory; +import org.apache.beam.sdk.testing.CoderProperties; +import org.apache.beam.sdk.util.SerializableUtils; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Throwables; +import org.joda.time.Duration; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; +import org.mockito.ArgumentCaptor; +import org.mockito.Mockito; + +/** Local unit tests for {@link JmsIO} that do not require an active JMS broker. */ +@RunWith(JUnit4.class) +public class JmsLocalTest { + + private static final String QUEUE = "queue"; + private static final String TOPIC = "topic"; + + @Test + public void testPipelineWithNonSerializableCF() { + SerializableUtils.ensureSerializable( + JmsIO.read() + .withConnectionFactoryProviderFn(__ -> new MockNonSerializableConnectionFactory())); + try { + SerializableUtils.ensureSerializable( + JmsIO.read().withConnectionFactory(new MockNonSerializableConnectionFactory())); + fail(); + } catch (Exception e) { + assertThat(Throwables.getRootCause(e), isA(NotSerializableException.class)); + } + } + + @Test + public void testSplitForQueue() throws Exception { + JmsIO.Read read = JmsIO.read().withQueue(QUEUE); + PipelineOptions pipelineOptions = PipelineOptionsFactory.create(); + int desiredNumSplits = 5; + JmsIO.UnboundedJmsSource initialSource = new JmsIO.UnboundedJmsSource<>(read); + List> splits = + initialSource.split(desiredNumSplits, pipelineOptions); + assertEquals(desiredNumSplits, splits.size()); + } + + @Test + public void testSplitForTopic() throws Exception { + JmsIO.Read read = JmsIO.read().withTopic(TOPIC); + PipelineOptions pipelineOptions = PipelineOptionsFactory.create(); + int desiredNumSplits = 5; + JmsIO.UnboundedJmsSource initialSource = new JmsIO.UnboundedJmsSource<>(read); + List> splits = + initialSource.split(desiredNumSplits, pipelineOptions); + assertEquals(1, splits.size()); + } + + @Test + public void testPublisherWithRetryConfiguration() { + RetryConfiguration retryPolicy = + RetryConfiguration.create(5, Duration.standardSeconds(15), null); + JmsIO.Write publisher = + JmsIO.write() + .withConnectionFactory(new ActiveMQConnectionFactory("vm://localhost")) + .withRetryConfiguration(retryPolicy) + .withQueue(QUEUE) + .withUsername("user") + .withPassword("password"); + assertEquals( + publisher.getRetryConfiguration(), + RetryConfiguration.create(5, Duration.standardSeconds(15), null)); + } + + @Test + public void testJmsCheckpointMarkIndividualAcknowledgeAllMessages() throws Exception { + Message msg1 = Mockito.mock(Message.class); + Message msg2 = Mockito.mock(Message.class); + Message msg3 = Mockito.mock(Message.class); + + JmsCheckpointMark.Preparer preparer = + JmsCheckpointMark.newPreparer(JmsIO.AcknowledgeMode.INDIVIDUAL_ACKNOWLEDGE); + preparer.add(msg1); + preparer.add(msg2); + preparer.add(msg3); + + AtomicInteger activeCheckpoints = new AtomicInteger(0); + JmsCheckpointMark mark = + preparer.newCheckpoint( + null, null, JmsIO.AcknowledgeMode.INDIVIDUAL_ACKNOWLEDGE, activeCheckpoints); + assertNotNull(mark.getMessages()); + assertEquals(3, mark.getMessages().size()); + assertNull(mark.getConsumer()); + assertNull(mark.getSession()); + assertEquals(1, activeCheckpoints.get()); + + mark.finalizeCheckpoint(); + + Mockito.verify(msg1, Mockito.times(1)).acknowledge(); + Mockito.verify(msg2, Mockito.times(1)).acknowledge(); + Mockito.verify(msg3, Mockito.times(1)).acknowledge(); + assertEquals(0, activeCheckpoints.get()); + } + + @Test + public void testJmsCheckpointMarkClientAcknowledgeUnsafeNoSessionRecreation() throws Exception { + Message msg1 = Mockito.mock(Message.class); + Message msg2 = Mockito.mock(Message.class); + + JmsCheckpointMark.Preparer preparer = + JmsCheckpointMark.newPreparer(JmsIO.AcknowledgeMode.CLIENT_ACKNOWLEDGE_UNSAFE); + preparer.add(msg1); + preparer.add(msg2); + + AtomicInteger activeCheckpoints = new AtomicInteger(0); + JmsCheckpointMark mark = + preparer.newCheckpoint( + null, null, JmsIO.AcknowledgeMode.CLIENT_ACKNOWLEDGE_UNSAFE, activeCheckpoints); + assertNotNull(mark.getMessages()); + assertEquals(1, mark.getMessages().size()); + assertNull(mark.getConsumer()); + assertNull(mark.getSession()); + assertEquals(1, activeCheckpoints.get()); + + mark.finalizeCheckpoint(); + + Mockito.verify(msg2, Mockito.times(1)).acknowledge(); + Mockito.verify(msg1, Mockito.never()).acknowledge(); + assertEquals(0, activeCheckpoints.get()); + } + + /** Test the checkpoint mark default coder, which is actually AvroCoder. */ + @Test + public void testCheckpointMarkDefaultCoder() throws Exception { + JmsCheckpointMark jmsCheckpointMark = + JmsCheckpointMark.newPreparer(JmsIO.AcknowledgeMode.CLIENT_ACKNOWLEDGE) + .newCheckpoint(null, null, JmsIO.AcknowledgeMode.CLIENT_ACKNOWLEDGE, null); + Coder coder = + new JmsIO.UnboundedJmsSource(null).getCheckpointMarkCoder(); + CoderProperties.coderSerializable(coder); + CoderProperties.coderDecodeEncodeEqual(coder, jmsCheckpointMark); + } + + @Test + public void testCloseWithTimeout() throws IOException, JMSException { + ConnectionFactory connectionFactory = Mockito.mock(ConnectionFactory.class); + Connection connection = Mockito.mock(Connection.class); + Session session = Mockito.mock(Session.class); + MessageConsumer consumer = Mockito.mock(MessageConsumer.class); + Queue queue = Mockito.mock(Queue.class); + + Mockito.when(connectionFactory.createConnection(Mockito.any(), Mockito.any())) + .thenReturn(connection); + Mockito.when(connection.createSession(Mockito.anyBoolean(), Mockito.anyInt())) + .thenReturn(session); + Mockito.when(session.createQueue(Mockito.anyString())).thenReturn(queue); + Mockito.when(session.createConsumer(Mockito.any())).thenReturn(consumer); + + Duration closeTimeout = Duration.millis(2000L); + JmsIO.Read spec = + JmsIO.read() + .withConnectionFactory(connectionFactory) + .withUsername("user") + .withPassword("password") + .withQueue(QUEUE) + .withCloseTimeout(closeTimeout); + + JmsIO.UnboundedJmsSource source = new JmsIO.UnboundedJmsSource<>(spec); + + ScheduledExecutorService mockScheduledExecutorService = + Mockito.mock(ScheduledExecutorService.class); + ExecutorOptions options = PipelineOptionsFactory.as(ExecutorOptions.class); + options.setScheduledExecutorService(mockScheduledExecutorService); + ArgumentCaptor runnableArgumentCaptor = ArgumentCaptor.forClass(Runnable.class); + Mockito.when( + mockScheduledExecutorService.schedule( + runnableArgumentCaptor.capture(), Mockito.anyLong(), Mockito.any(TimeUnit.class))) + .thenReturn(null /* unused */); + + JmsIO.UnboundedJmsReader reader = source.createReader(options, null); + reader.start(); + assertFalse(getDiscardedValue(reader)); + reader.checkpointMarkPreparer.add(Mockito.mock(Message.class)); + org.apache.beam.sdk.io.UnboundedSource.CheckpointMark mark = reader.getCheckpointMark(); + reader.close(); + assertTrue(getDiscardedValue(reader)); + Mockito.verify(mockScheduledExecutorService) + .schedule(Mockito.any(Runnable.class), Mockito.eq(1L), Mockito.eq(TimeUnit.SECONDS)); + mark.finalizeCheckpoint(); + runnableArgumentCaptor.getValue().run(); + assertTrue(getDiscardedValue(reader)); + Mockito.verifyNoMoreInteractions(mockScheduledExecutorService); + } + + private boolean getDiscardedValue(JmsIO.UnboundedJmsReader reader) { + JmsCheckpointMark.Preparer preparer = reader.checkpointMarkPreparer; + preparer.lock.readLock().lock(); + try { + return preparer.discarded; + } finally { + preparer.lock.readLock().unlock(); + } + } +} diff --git a/sdks/python/apache_beam/runners/dataflow/internal/apiclient.py b/sdks/python/apache_beam/runners/dataflow/internal/apiclient.py index ac4118643109..0875bdd14df5 100644 --- a/sdks/python/apache_beam/runners/dataflow/internal/apiclient.py +++ b/sdks/python/apache_beam/runners/dataflow/internal/apiclient.py @@ -280,8 +280,10 @@ def __init__( for k, v in sdk_pipeline_options.items(): if v is None: continue - options_dict[k] = str(v) if isinstance( - v, value_provider.ValueProvider) else v + if isinstance(v, value_provider.ValueProvider): + options_dict[k] = v.get() if v.is_accessible() else None + else: + options_dict[k] = v options_dict["pipelineUrl"] = proto_pipeline_staged_url if pipeline_proto_hash: options_dict["pipelineProtoHash"] = pipeline_proto_hash diff --git a/sdks/python/apache_beam/runners/dataflow/internal/apiclient_test.py b/sdks/python/apache_beam/runners/dataflow/internal/apiclient_test.py index dc55a28cecf4..4fca13abee99 100644 --- a/sdks/python/apache_beam/runners/dataflow/internal/apiclient_test.py +++ b/sdks/python/apache_beam/runners/dataflow/internal/apiclient_test.py @@ -113,6 +113,25 @@ def test_pipeline_url(self): self.assertEqual(pipeline_url, FAKE_PIPELINE_URL) + def test_value_provider_options_serialization(self): + class UserOptions(PipelineOptions): + @classmethod + def _add_argparse_args(cls, parser): + parser.add_value_provider_argument('--at_vp_arg1') + parser.add_value_provider_argument('--at_vp_arg2') + + pipeline_options = UserOptions([ + '--at_vp_arg2', 'provided', '--temp_location', 'gs://any-location/temp' + ]) + env = apiclient.Environment([], + pipeline_options, + '2.0.0', + FAKE_PIPELINE_URL) + + recovered_options = env.proto.sdk_pipeline_options['options'] + self.assertIsNone(recovered_options['at_vp_arg1']) + self.assertEqual(recovered_options['at_vp_arg2'], 'provided') + def test_pipeline_proto_hash(self): pipeline_options = PipelineOptions( ['--temp_location', 'gs://any-location/temp']) diff --git a/sdks/python/test-suites/direct/build.gradle b/sdks/python/test-suites/direct/build.gradle index d1fe45683a83..2c71c81afaa9 100644 --- a/sdks/python/test-suites/direct/build.gradle +++ b/sdks/python/test-suites/direct/build.gradle @@ -44,7 +44,8 @@ task ioCrossLanguagePostCommit { } task messagingCrossLanguagePostCommit { - getVersionsAsList('cross_language_validates_py_versions').each { + // Messaging E2E tests has testcontainers overhead. Single Python version suffices and reducing CI flakiness + getVersionsAsList('cross_language_validates_py_versions').take(1).each { dependsOn.add(":sdks:python:test-suites:direct:py${getVersionSuffix(it)}:messagingCrossLanguagePythonUsingJava") } } diff --git a/website/www/site/content/en/documentation/io/connectors.md b/website/www/site/content/en/documentation/io/connectors.md index 242a255ba82d..679b6bf1e0e3 100644 --- a/website/www/site/content/en/documentation/io/connectors.md +++ b/website/www/site/content/en/documentation/io/connectors.md @@ -445,7 +445,10 @@ This table provides a consolidated, at-a-glance overview of the available built- ✔ native - Not available + + ✔ + via X-language + Not available Not available Not available @@ -1269,7 +1272,10 @@ This table provides a consolidated, at-a-glance overview of the available built- ✔ native - Not available + + ✔ + via X-language + Not available Not available