Fix Illegal to use destination for jms producer

[Abacn]

Fixes #25849

The issue is that the producer was created with a destination in #24973. This kind of producer cannot set destination again in send()

Please add a meaningful description for your change here

---

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
• Update CHANGES.md with noteworthy changes.
• If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI.

[Abacn]

R: @sriram23kmm
CC: @Amraneze

This indicates we need some integration test for JmsIO (entered #25887). Would help if you are interested in adding integration tests.

[github-actions]

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control

[Amraneze]

Hey @Abacn, We got the same issue in our pipeline too. We added some workaround. But, I will create a new PR to add more IT tests.

[Abacn]

Hi @Amraneze, thanks for the work. Is this small fix work or there are other things need fix? If this suffices we can get this in and have a nightly snapshot version that can be used as a workaround.

[Amraneze]

There are two things with Apache Qpid connector, we can't use static topic. The workaround was to use dynamic topic and give a function that returns always the same value. Also, with the same connector, we found out that the number of ports opened is around 5k (18 workers * 300) which consume the majority of the ports in our NAT gateway. But, using the same code of JmsIO without the new version of Apache beam works fine.

Note: We don't want to use the option numberOfWorkerHarnessThreads

[Amraneze]

For the IT tests with other connectors, we don't have access to the library TIBCO. What should we do ?

[Abacn]

There are two things with Apache Qpid connector, we can't use static topic. The workaround was to use dynamic topic and give a function that returns always the same value. Also, with the same connector, we found out that the number of ports opened is around 5k (18 workers * 300) which consume the majority of the ports in our NAT gateway. But, using the same code of JmsIO without the new version of Apache beam works fine.

Note: We don't want to use the option numberOfWorkerHarnessThreads

What is 300 coming from? Is it the number of thread per worker, or total number of bundles? This sounds like connection leaking.

Also, what is the timeline for the fix? If not before the cut of 2.47 (beginning of April we should revert #24973 at the moment

[Amraneze]

What is 300 coming from? Is it the number of thread per worker, or total number of bundles? This sounds like connection leaking.

For now, we are testing with numberOfWorkerHarnessThreads to confirm the hypothesis that it's related to number of thread per worker. I will let you know about it.

Also, what is the timeline for the fix? If not before the cut of 2.47 (beginning of April we should revert #25887 at the moment

The problem is that we tested the same code (JmsIO) that we merged in our pipelines with the version 2.45 and we didn't have any issues (for the last 2 weeks). It's just when we upgraded the apache beam to 2.46, so I'm not sure if it's related to JmsIO.

[Abacn]

@Amraneze is both pipeline (with/without excessive port occupying) running on same version of Dataflow runner (v1 or v2) ?

Anyway this is another issue than that reported in #25849. @sriram23kmm are you able to try the fix of this PR, that is compile your own jmsio artifact and replace that shipped with Beam v2.46.0 to validate it fixes your issue?

[Amraneze]

@Abacn It's the same pipeline one was running with the version 2.45 + the new code for JmsIO and one was with the version 2.46. Both of them have the same configuration (same number of workers, ....) and they use Dataflow Runner v1 (we had some drawbacks with v2)

Here is a screenshot of the port usage after running the pipeline (using only beam 2.46) with numberOfWorkerHarnessThreads = 32. It stabilizes at 2k open ports.

A screenshot of the port usage after running the pipeline (using beam 2.46 + the fix of this PR). It stabilizes at 2k open ports.

A screenshot of the port usage after running the pipeline (using beam 2.45 + the fix of this PR + new JmsIO code). It stabilizes at 2k open ports.

Note: The pipelines are running with the same configuration, 18 workers and n2-highmem-4 without autoscaling

I guess, we will need to setup the connection at the setup cycle instead of startBundle cycle

[sriram23kmm]

@Abacn I tried this PR fix + 2.46 beam it solves our problem.

Also, when it is going release in the next version(2.47.0)?

[Abacn]

@sriram23kmm that's great. Let us get this in first then. The port usage is another issue than the original bug and could be fixed separately. The next version v2.47.0 which is expected be released in late April.

[sriram23kmm]

Thanks, @Abacn so then can you merge this PR and get this immediately since it was affecting our production

[Abacn]

done. The nightly snapshot is in the directory https://repository.apache.org/content/repositories/snapshots/org/apache/beam/beam-sdks-java-io-jms/2.47.0-SNAPSHOT/ and will be updated to include the fix in the next snapshot.