[#27312] fix(JmsIO): create a session pool for JmsIO #27312 #27313

Amraneze · 2023-06-30T07:54:02Z

[#26203] fix(JmsIO): use a unused port for tests in JmsIO #26203

In this PR, I added a session pool so that it can be used for either reading or writing (for now it's only for writing). This will allow the user to configure the number of connection maximum to not be exceeded.

Note: We used Apache Commons Pool, but they are some module that are more performant.

https://github.com/DanielYWoo/fast-object-pool

https://code.google.com/archive/p/furious-objectpool/

https://github.com/chrisvest/stormpot/
Here is an article that display the benchmark of each object pool

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.

[apache#26203] fix(JmsIO): use a unused port for tests in JmsIO apache#26203 Fixes apache#27312, apache#26203 Co-Authored-By: Amrane Ait Zeouay <[email protected]>

codecov · 2023-06-30T08:22:31Z

Codecov Report

Merging #27313 (55cb703) into master (04fdf08) will increase coverage by 0.02%.
The diff coverage is n/a.

@@            Coverage Diff             @@
##           master   #27313      +/-   ##
==========================================
+ Coverage   71.49%   71.51%   +0.02%     
==========================================
  Files         858      858              
  Lines      104809   104809              
==========================================
+ Hits        74935    74958      +23     
+ Misses      28326    28303      -23     
  Partials     1548     1548

Flag	Coverage Δ
python	`80.85% <ø> (+0.03%)`	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

see 10 files with indirect coverage changes

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more

github-actions · 2023-06-30T08:35:44Z

Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment assign set of reviewers

damccorm · 2023-07-11T17:35:04Z

@johnjcasey could you take a look at this one?

johnjcasey · 2023-07-11T17:45:04Z

sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsIO.java

@@ -699,22 +701,24 @@ protected void finalize() {
  public abstract static class Write<EventT>
      extends PTransform<PCollection<EventT>, WriteJmsResult<EventT>> {

-    abstract @Nullable ConnectionFactory getConnectionFactory();
+    public abstract @Nullable ConnectionFactory getConnectionFactory();


why are these public?

If I still remember, I was trying to access to it from the new package pool. We can move the pool classes to the main package to keep all functions as they were

johnjcasey · 2023-07-11T17:51:15Z

sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/pool/JmsSessionFactory.java

+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@SuppressWarnings({


We should not include new blanket nullness supression

I will remove it

johnjcasey · 2023-07-11T17:55:54Z

sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/pool/JmsSessionFactory.java

+    return new DefaultPooledObject<>(session);
+  }
+
+  private <U> U callSessionMethod(Session session, String methodName, U defaultValue) {


Why are we doing this via reflection?

Because the Javax API doesn't provide any function for closing the session/connection. But, the two JMS clients (ActiveMQ & qpid) have the same functions getSession & getConnection. The main problem that I found out is that the connection is not getting closed after the session is closed.

Note: I didn't check other clients

I'm using JmsIO with the Solace broker and this reflection usage will most definitely not work with the JCSMP API which Solace uses to implement JMS and probably will break JmsIO usage with Solace altogether... so please consider that checking two clients might not be enough...

@ppawel which client are using with the connector JmsIO ? We are using Solace as a message broker too but with Qpid as a client (The connector Solace is not well maintained). It would be great to add the client that you are using in the test cases so we won't make sure we don't have any regression

I am using the official Solace implementation of JMS (sol-jms) but it is giving us a lot of headaches combined with Apache Beam streaming. I might try the Qpid JMS client, but in any case I am looking forward to this PR being integrated as well because currently one of the problems we observe is connection count blowing up under load.

johnjcasey · 2023-07-11T18:00:55Z

sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/pool/SerializableSessionPool.java

+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.commons.pool2.ObjectPool;
+
+public abstract class SerializableSessionPool<T> implements ObjectPool<T>, Serializable, Closeable {


Can we add some tests for this session pool in specific? It seems like there is a theoretical risk of incorrectly borrowing/returning objects

I did some tests using a Docker image of SolaceUI and closing the connection manually. Everything works fine. I tried to add some Unit Tests but I was blocked. I was thinking to pass a SessionPoolFactory so that it makes the tests easier to write. What do you think?

Note: I couldn't run the pipeline and stop the session, because I don't have control over that and also using DirectRunner I couldn't create multiple threads to test if the session can be borrowed after reaching the maximum limit of sessions

I'm all for writing code that is easy to test

aromanenko-dev · 2023-08-01T16:13:42Z

@Amraneze Kind ping on this PR

Amraneze · 2023-08-02T19:56:32Z

@Amraneze Kind ping on this PR

Sorry, I was on holidays. I will find time for it this week or next week to work on it

github-actions · 2023-10-02T12:41:27Z

This pull request has been marked as stale due to 60 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the [email protected] list. Thank you for your contributions.

github-actions · 2023-10-09T12:42:31Z

This pull request has been closed due to lack of activity. If you think that is incorrect, or the pull request requires review, you can revive the PR at any time.

[apache#27312] fix(JmsIO): create a session pool for JmsIO apache#27312

55cb703

[apache#26203] fix(JmsIO): use a unused port for tests in JmsIO apache#26203 Fixes apache#27312, apache#26203 Co-Authored-By: Amrane Ait Zeouay <[email protected]>

github-actions bot added java build io jdbc jms labels Jun 30, 2023

johnjcasey requested changes Jul 11, 2023

View reviewed changes

github-actions bot added the stale label Oct 2, 2023

github-actions bot closed this Oct 9, 2023

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[#27312] fix(JmsIO): create a session pool for JmsIO #27312 #27313

[#27312] fix(JmsIO): create a session pool for JmsIO #27312 #27313

Amraneze commented Jun 30, 2023

codecov bot commented Jun 30, 2023 •

edited

Loading

github-actions bot commented Jun 30, 2023

damccorm commented Jul 11, 2023

johnjcasey Jul 11, 2023

Amraneze Jul 11, 2023

johnjcasey Jul 11, 2023

Amraneze Jul 11, 2023

johnjcasey Jul 11, 2023

Amraneze Jul 11, 2023 •

edited

Loading

ppawel Jul 14, 2023

Amraneze Jul 15, 2023

ppawel Jul 16, 2023

johnjcasey Jul 11, 2023

Amraneze Jul 11, 2023 •

edited

Loading

johnjcasey Jul 19, 2023 •

edited

Loading

aromanenko-dev commented Aug 1, 2023

Amraneze commented Aug 2, 2023

github-actions bot commented Oct 2, 2023

github-actions bot commented Oct 9, 2023

[#27312] fix(JmsIO): create a session pool for JmsIO #27312 #27313

[#27312] fix(JmsIO): create a session pool for JmsIO #27312 #27313

Conversation

Amraneze commented Jun 30, 2023

GitHub Actions Tests Status (on master branch)

codecov bot commented Jun 30, 2023 • edited Loading

Codecov Report

github-actions bot commented Jun 30, 2023

damccorm commented Jul 11, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Amraneze Jul 11, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Amraneze Jul 11, 2023 • edited Loading

Choose a reason for hiding this comment

johnjcasey Jul 19, 2023 • edited Loading

Choose a reason for hiding this comment

aromanenko-dev commented Aug 1, 2023

Amraneze commented Aug 2, 2023

github-actions bot commented Oct 2, 2023

github-actions bot commented Oct 9, 2023

codecov bot commented Jun 30, 2023 •

edited

Loading

Amraneze Jul 11, 2023 •

edited

Loading

Amraneze Jul 11, 2023 •

edited

Loading

johnjcasey Jul 19, 2023 •

edited

Loading