JMSStreamer enhancement -- needs to support multiple IgniteDataStreamers.

[Timothy Steffens]

Dev -

I have a use case that is not able to be addressed with the existing JMSStreamer functionality so I would like to propose an enhancement.

The use case is such that I have a series of database tables and am writing messages into a shared JMS queue for each table update. These messages are then picked up by the JMSStreamer, converted into various object types (based on an indicator within the message), and written into a shared Ignite cache that holds all object types. I would like to enhance the JMSStreamer so that it accepts multiple IgniteDataStreamer instances so that it can stream each object type into its own corresponding cache rather than having to stream into a shared, generic cache.

My approach will be to take advantage of the inherited StreamAdapter functionality within the JMSStreamer. Note that my changes will be backwards compatible so that other streamers are not broken. Additionally, these changes will streamline the message processing functionality which is duplicated between the StreamAdapter and JMSStreamer today.


StreamAdapter  
  - Update the StreamAdapter to accept a list of IgniteDataStreamers - the getStreamer() method will be kept for backwards compatibility and I will add getStreamers().
  
  - The StreamAdapter's addMessage() method now has to interrogate the Message type to determine onto which streamer to place the message. This can be done by providing a custom map that relates the message type to a specific streamer.


JMSStreamer  
  - Update the JMSStreamer add a setStreamers() method that takes a list of streamers - the setStreamer() method will be left for backwards compatibility.
  
  - The start() method is updated to verify the list of streamers is not null and to verify that a Single/MultipleTupleExtractor was provided instead of a MessageTransformer.
  
  - Mark the setTransformer() method as deprecated.
  
  - The executor.execute() method is updated to use the inherited addMessage() method from StreamAdapter.
  
  - Delete the JMSStreamer private processMethod() after refactoring to use the StreamAdapter addMessage() method.
  
  - Remove the MessageTransformer property from the JMSStreamer because it will now use the Single/MultipleTupleExtractor within the StreamAdapter - these two classes do basically the same thing. The MessageTransformer class can then be deprecated and go away as it is not used anywhere else in the code base.


MessageTransformer  
  - Mark as deprecated.



Please review and let me know how to proceed.

Thank you.

[dsetrakyan]

Tim,

I am not sure why we need to change the JMS streamer. Why not just create a
separate JMS streamer per cache? In my view, it is more intuitive to keep
one-to-one relationship between JMS streamer and the data streamer.

D.

On Wed, Nov 8, 2017 at 9:40 AM, Timothy Steffens <
[hidden email]> wrote:

> Dev -
>
> I have a use case that is not able to be addressed with the existing
> JMSStreamer functionality so I would like to propose an enhancement.
>
> The use case is such that I have a series of database tables and am
> writing messages into a shared JMS queue for each table update. These
> messages are then picked up by the JMSStreamer, converted into various
> object types (based on an indicator within the message), and written into a
> shared Ignite cache that holds all object types. I would like to enhance
> the JMSStreamer so that it accepts multiple IgniteDataStreamer instances so
> that it can stream each object type into its own corresponding cache rather
> than having to stream into a shared, generic cache.
>
> My approach will be to take advantage of the inherited StreamAdapter
> functionality within the JMSStreamer. Note that my changes will be
> backwards compatible so that other streamers are not broken. Additionally,
> these changes will streamline the message processing functionality which is
> duplicated between the StreamAdapter and JMSStreamer today.
>
>
> StreamAdapter
>   - Update the StreamAdapter to accept a list of IgniteDataStreamers - the
> getStreamer() method will be kept for backwards compatibility and I will
> add getStreamers().
>
>   - The StreamAdapter's addMessage() method now has to interrogate the
> Message type to determine onto which streamer to place the message. This
> can be done by providing a custom map that relates the message type to a
> specific streamer.
>
>
> JMSStreamer
>   - Update the JMSStreamer add a setStreamers() method that takes a list
> of streamers - the setStreamer() method will be left for backwards
> compatibility.
>
>   - The start() method is updated to verify the list of streamers is not
> null and to verify that a Single/MultipleTupleExtractor was provided
> instead of a MessageTransformer.
>
>   - Mark the setTransformer() method as deprecated.
>
>   - The executor.execute() method is updated to use the inherited
> addMessage() method from StreamAdapter.
>
>   - Delete the JMSStreamer private processMethod() after refactoring to
> use the StreamAdapter addMessage() method.
>
>   - Remove the MessageTransformer property from the JMSStreamer because it
> will now use the Single/MultipleTupleExtractor within the StreamAdapter -
> these two classes do basically the same thing. The MessageTransformer class
> can then be deprecated and go away as it is not used anywhere else in the
> code base.
>
>
> MessageTransformer
>   - Mark as deprecated.
>
>
>
> Please review and let me know how to proceed.
>
> Thank you.
>

[Timothy Steffens]

 Can the JMSStreamer be configured to inspect the message prior to picking it off of the queue? If so, then I can use multiple JMSStreamers as you suggest. If not, then I cannot since each JMSStreamer will not know if the next message is for them or for another to process. I do not want to have multiple queues (one queue per table - which would equate to one JMSStreamer per queue) due to ongoing maintenance concerns.

    On Wednesday, November 8, 2017, 1:45:33 PM CST, Dmitriy Setrakyan <[hidden email]> wrote:  
 
 Tim,

I am not sure why we need to change the JMS streamer. Why not just create a
separate JMS streamer per cache? In my view, it is more intuitive to keep
one-to-one relationship between JMS streamer and the data streamer.

D.

On Wed, Nov 8, 2017 at 9:40 AM, Timothy Steffens <
[hidden email]> wrote:

> Dev -
>
> I have a use case that is not able to be addressed with the existing
> JMSStreamer functionality so I would like to propose an enhancement.
>
> The use case is such that I have a series of database tables and am
> writing messages into a shared JMS queue for each table update. These
> messages are then picked up by the JMSStreamer, converted into various
> object types (based on an indicator within the message), and written into a
> shared Ignite cache that holds all object types. I would like to enhance
> the JMSStreamer so that it accepts multiple IgniteDataStreamer instances so
> that it can stream each object type into its own corresponding cache rather
> than having to stream into a shared, generic cache.
>
> My approach will be to take advantage of the inherited StreamAdapter
> functionality within the JMSStreamer. Note that my changes will be
> backwards compatible so that other streamers are not broken. Additionally,
> these changes will streamline the message processing functionality which is
> duplicated between the StreamAdapter and JMSStreamer today.
>
>
> StreamAdapter
>  - Update the StreamAdapter to accept a list of IgniteDataStreamers - the
> getStreamer() method will be kept for backwards compatibility and I will
> add getStreamers().
>
>  - The StreamAdapter's addMessage() method now has to interrogate the
> Message type to determine onto which streamer to place the message. This
> can be done by providing a custom map that relates the message type to a
> specific streamer.
>
>
> JMSStreamer
>  - Update the JMSStreamer add a setStreamers() method that takes a list
> of streamers - the setStreamer() method will be left for backwards
> compatibility.
>
>  - The start() method is updated to verify the list of streamers is not
> null and to verify that a Single/MultipleTupleExtractor was provided
> instead of a MessageTransformer.
>
>  - Mark the setTransformer() method as deprecated.
>
>  - The executor.execute() method is updated to use the inherited
> addMessage() method from StreamAdapter.
>
>  - Delete the JMSStreamer private processMethod() after refactoring to
> use the StreamAdapter addMessage() method.
>
>  - Remove the MessageTransformer property from the JMSStreamer because it
> will now use the Single/MultipleTupleExtractor within the StreamAdapter -
> these two classes do basically the same thing. The MessageTransformer class
> can then be deprecated and go away as it is not used anywhere else in the
> code base.
>
>
> MessageTransformer
>  - Mark as deprecated.
>
>
>
> Please review and let me know how to proceed.
>
> Thank you.
>

 

[dsetrakyan]

On Thu, Nov 9, 2017 at 5:06 AM, Timothy Steffens <[hidden email]
> wrote:

> Can the JMSStreamer be configured to inspect the message prior to picking
> it off of the queue? If so, then I can use multiple JMSStreamers as you
> suggest. If not, then I cannot since each JMSStreamer will not know if the
> next message is for them or for another to process. I do not want to have
> multiple queues (one queue per table - which would equate to one
> JMSStreamer per queue) due to ongoing maintenance concerns.
>

To be honest, queue-per-table seems like the most appropriate solution
here. I do not think inspecting messages before they get picked off the
queue will perform well.

If queue-per-table is not possible, then my suggestion would be not to use
JMS Streamer at all, and implement your own logic for reading JMS messages
and passing them off to the appropriate data streamer.

D.

[Valentin Kulichenko]

Tim,

I think you properly defined flaws in the JmsStreamer and it defintely
makes sense to do the following:

   - Deprecate org.apache.ignite.stream.jms11.MessageTransformer in favor
   of extractors defined in StreamAdapter. This JMS specific transformer
   doesn't provide any value and only duplicates generic extractors making it
   very confusing.
   - Remove JmsStreamer#processMessage method and use
   StreamAdapter#addMessage.

This will make both API and implementation cleaner and more in line with
overall streamers design.

And the best thing: there is no need to add special support for multiple
streamers anymore. If one has one queue/many caches use case, they can
simply extend JmsStreamer and override addMessage method (which is
protected). This way the behavior can be modified in any desired way, while
reusing most of JmsStreamer code.

Thoughts?

-Val

On Wed, Nov 8, 2017 at 1:31 PM, Dmitriy Setrakyan <[hidden email]>
wrote:

> On Thu, Nov 9, 2017 at 5:06 AM, Timothy Steffens <
> [hidden email]
> > wrote:
>
> > Can the JMSStreamer be configured to inspect the message prior to picking
> > it off of the queue? If so, then I can use multiple JMSStreamers as you
> > suggest. If not, then I cannot since each JMSStreamer will not know if
> the
> > next message is for them or for another to process. I do not want to have
> > multiple queues (one queue per table - which would equate to one
> > JMSStreamer per queue) due to ongoing maintenance concerns.
> >
>
> To be honest, queue-per-table seems like the most appropriate solution
> here. I do not think inspecting messages before they get picked off the
> queue will perform well.
>
> If queue-per-table is not possible, then my suggestion would be not to use
> JMS Streamer at all, and implement your own logic for reading JMS messages
> and passing them off to the appropriate data streamer.
>
> D.
>

[Timothy Steffens]

 Yes, this makes sense. The proposed changes will improve the overall implementation of the JMSStreamer and the changes will also allow for me to then extend the streamer and implement my own custom functionality as needed.
What are the next steps to getting this change implemented?


    On Thursday, November 9, 2017, 6:11:01 PM CST, Valentin Kulichenko <[hidden email]> wrote:  
 
 Tim,

I think you properly defined flaws in the JmsStreamer and it defintely
makes sense to do the following:

  - Deprecate org.apache.ignite.stream.jms11.MessageTransformer in favor
  of extractors defined in StreamAdapter. This JMS specific transformer
  doesn't provide any value and only duplicates generic extractors making it
  very confusing.
  - Remove JmsStreamer#processMessage method and use
  StreamAdapter#addMessage.

This will make both API and implementation cleaner and more in line with
overall streamers design.

And the best thing: there is no need to add special support for multiple
streamers anymore. If one has one queue/many caches use case, they can
simply extend JmsStreamer and override addMessage method (which is
protected). This way the behavior can be modified in any desired way, while
reusing most of JmsStreamer code.

Thoughts?

-Val

On Wed, Nov 8, 2017 at 1:31 PM, Dmitriy Setrakyan <[hidden email]>
wrote:

> On Thu, Nov 9, 2017 at 5:06 AM, Timothy Steffens <
> [hidden email]
> > wrote:
>
> > Can the JMSStreamer be configured to inspect the message prior to picking
> > it off of the queue? If so, then I can use multiple JMSStreamers as you
> > suggest. If not, then I cannot since each JMSStreamer will not know if
> the
> > next message is for them or for another to process. I do not want to have
> > multiple queues (one queue per table - which would equate to one
> > JMSStreamer per queue) due to ongoing maintenance concerns.
> >
>
> To be honest, queue-per-table seems like the most appropriate solution
> here. I do not think inspecting messages before they get picked off the
> queue will perform well.
>
> If queue-per-table is not possible, then my suggestion would be not to use
> JMS Streamer at all, and implement your own logic for reading JMS messages
> and passing them off to the appropriate data streamer.
>
> D.
>

 

[Valentin Kulichenko]

Tim,

If you want to do this change by yourself, then create a ticket in Jira,
assign to yourself and prepare a pull request.

See here for more information about the process:
https://ignite.apache.org/community/contribute.html#contribute
https://cwiki.apache.org/confluence/display/IGNITE/How+to+Contribute

-Val

On Thu, Nov 9, 2017 at 8:34 PM Timothy Steffens <[hidden email]>
wrote:

> Yes, this makes sense. The proposed changes will improve the overall
> implementation of the JMSStreamer and the changes will also allow for me to
> then extend the streamer and implement my own custom functionality as
> needed.
>
> What are the next steps to getting this change implemented?
>
>
>
> On Thursday, November 9, 2017, 6:11:01 PM CST, Valentin Kulichenko <
> [hidden email]> wrote:
>
>
> Tim,
>
> I think you properly defined flaws in the JmsStreamer and it defintely
> makes sense to do the following:
>
>   - Deprecate org.apache.ignite.stream.jms11.MessageTransformer in favor
>
>   of extractors defined in StreamAdapter. This JMS specific transformer
>   doesn't provide any value and only duplicates generic extractors making
> it
>   very confusing.
>   - Remove JmsStreamer#processMessage method and use
>
>   StreamAdapter#addMessage.
>
> This will make both API and implementation cleaner and more in line with
> overall streamers design.
>
> And the best thing: there is no need to add special support for multiple
> streamers anymore. If one has one queue/many caches use case, they can
> simply extend JmsStreamer and override addMessage method (which is
> protected). This way the behavior can be modified in any desired way, while
> reusing most of JmsStreamer code.
>
> Thoughts?
>
> -Val
>
> On Wed, Nov 8, 2017 at 1:31 PM, Dmitriy Setrakyan <[hidden email]>
> wrote:
>
> > On Thu, Nov 9, 2017 at 5:06 AM, Timothy Steffens <
> > [hidden email]
> > > wrote:
> >
> > > Can the JMSStreamer be configured to inspect the message prior to
> picking
> > > it off of the queue? If so, then I can use multiple JMSStreamers as you
> > > suggest. If not, then I cannot since each JMSStreamer will not know if
> > the
> > > next message is for them or for another to process. I do not want to
> have
> > > multiple queues (one queue per table - which would equate to one
> > > JMSStreamer per queue) due to ongoing maintenance concerns.
> > >
> >
> > To be honest, queue-per-table seems like the most appropriate solution
> > here. I do not think inspecting messages before they get picked off the
> > queue will perform well.
> >
> > If queue-per-table is not possible, then my suggestion would be not to
> use
> > JMS Streamer at all, and implement your own logic for reading JMS
> messages
> > and passing them off to the appropriate data streamer.
> >
> > D.
> >
>