[GitHub] [flink] senegalo commented on pull request #12056: [FLINK-17502] [flink-connector-rabbitmq] RMQSource refactor

[GitBox]

senegalo commented on pull request #12056:
URL: https://github.com/apache/flink/pull/12056#issuecomment-708418856


   @dawidwys i did a bit of changes today i'll try and do more tomorrow if you 
beat me to the punch so be it :)
   
   i plugged in your changes but the tests are red and there were a few tweaks 
that still needed to be done. I'll try and work on that tomorrow or the day 
after.



This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [flink] senegalo commented on pull request #12056: [FLINK-17502] [flink-connector-rabbitmq] RMQSource refactor

[GitBox]

senegalo commented on pull request #12056:
URL: https://github.com/apache/flink/pull/12056#issuecomment-707626966


   Hey @dawidwys I am really sorry was away all weekend and couldn't address 
your issues... Also sorry about the silly issues here and there "my experience 
in Java is measured in weeks !" 
   Absolutely no problem if you want to take over the remaining issues. If you 
don't have time i was going to try and tackle them this week / weekend.



This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [flink] senegalo commented on pull request #12056: [FLINK-17502] [flink-connector-rabbitmq] RMQSource refactor

[GitBox]

senegalo commented on pull request #12056:
URL: https://github.com/apache/flink/pull/12056#issuecomment-703233062


   @dawidwys i took care of every single comment you mentioned hope that does 
the deed.



This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [flink] senegalo commented on pull request #12056: [FLINK-17502] [flink-connector-rabbitmq] RMQSource refactor

[GitBox]

senegalo commented on pull request #12056:
URL: https://github.com/apache/flink/pull/12056#issuecomment-703232999


   @flinkbot run azure



This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [flink] senegalo commented on pull request #12056: [FLINK-17502] [flink-connector-rabbitmq] RMQSource refactor

[GitBox]

senegalo commented on pull request #12056:
URL: https://github.com/apache/flink/pull/12056#issuecomment-702037986


   @flinkbot run azure



This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [flink] senegalo commented on pull request #12056: [FLINK-17502] [flink-connector-rabbitmq] RMQSource refactor

[GitBox]

senegalo commented on pull request #12056:
URL: https://github.com/apache/flink/pull/12056#issuecomment-701879609


   @flinkbot run azure



This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [flink] senegalo commented on pull request #12056: [FLINK-17502] [flink-connector-rabbitmq] RMQSource refactor

[GitBox]

senegalo commented on pull request #12056:
URL: https://github.com/apache/flink/pull/12056#issuecomment-701266601


   @dawidwys Ohh .. i didn't know that ! anyway i just created a local branch 
from master with the history and squash merged everything since i couldn't 
rebase without merging line by line.
   
   Let me know if there is anything else i can do
   
   Thanks for your time :)



This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [flink] senegalo commented on pull request #12056: [FLINK-17502] [flink-connector-rabbitmq] RMQSource refactor

[GitBox]

senegalo commented on pull request #12056:
URL: https://github.com/apache/flink/pull/12056#issuecomment-701257642


   heemm .. maybe it's only in github enterprise then.
   Okay then i'll try to rebase one more time .. if it takes too much time i'll 
squash them. 



This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [flink] senegalo commented on pull request #12056: [FLINK-17502] [flink-connector-rabbitmq] RMQSource refactor

[GitBox]

senegalo commented on pull request #12056:
URL: https://github.com/apache/flink/pull/12056#issuecomment-701253493


   In github there is the option "when the PR is approved" to squash merge the 
changes instead of the normal merge. Usually the merge button has an arrow 
beneath it that lets you select between both. I honestly thought that was going 
to be the case at the end since all the commits in the logs neatly match the PR 
names. 
   So can we do that as rebasing everything to master is not straight forward 
at all and i would like to keep the history till we close the PR.
   Let me know if this is an option or not otherwise I'll rebase 
   



This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [flink] senegalo commented on pull request #12056: [FLINK-17502] [flink-connector-rabbitmq] RMQSource refactor

[GitBox]

senegalo commented on pull request #12056:
URL: https://github.com/apache/flink/pull/12056#issuecomment-701216291


   @dawidwys you mean only the last one ? 
   because i did that several times before
   otherwise i would just reset the branch and cherry pick my commits again 



This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [flink] senegalo commented on pull request #12056: [FLINK-17502] [flink-connector-rabbitmq] RMQSource refactor

[GitBox]

senegalo commented on pull request #12056:
URL: https://github.com/apache/flink/pull/12056#issuecomment-701181655


   @flinkbot run azure



This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [flink] senegalo commented on pull request #12056: [FLINK-17502] [flink-connector-rabbitmq] RMQSource refactor

[GitBox]

senegalo commented on pull request #12056:
URL: https://github.com/apache/flink/pull/12056#issuecomment-700850674


   Sure will do that right away :)
   
   On Tue 29. Sep 2020 at 19:02, Dawid Wysakowicz 
   wrote:
   
   >
   >
   > Sorry, unfortunately you are right that I forgot about it :(. I will try
   > to have a look this week. Could you maybe rebase it on top of the current
   > master in the meantime?
   >
   >
   >
   >
   > —
   > You are receiving this because you were mentioned.
   > Reply to this email directly, view it on GitHub
   > , or
   > unsubscribe
   > 

   > .
   >
   >
   >
   



This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [flink] senegalo commented on pull request #12056: [FLINK-17502] [flink-connector-rabbitmq] RMQSource refactor

[GitBox]

senegalo commented on pull request #12056:
URL: https://github.com/apache/flink/pull/12056#issuecomment-699868484


   Hey guys did @austince  or @dawidwys had any time to look at this ? it's 
been so long in the making that i think you guys completely forgot about it !



This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [flink] senegalo commented on pull request #12056: [FLINK-17502] [flink-connector-rabbitmq] RMQSource refactor

[GitBox]

senegalo commented on pull request #12056:
URL: https://github.com/apache/flink/pull/12056#issuecomment-685341691


   @austince @dawidwys
   Hey guys, 
   Finally got around doing the changes you suggested.
   
   Now we do have a reset message "that was obviously needed for some reason i 
thought we call the run function with every message !!! which would've create a 
new instance of the collector that's why i didn't understand why we would need 
the reset function... sometimes i just amaze myself !"
   
   As for the checks, currently the user has to call the 
`setMessageIdentifiers` so set the correlationId and deliveryTag which if the 
checks pass would set the `messageValidated` flag to true when that happens the 
`collect` message accepts and appends the message to the context if not it 
ignores it if it's `null` it throws an error telling the user that it must be 
called at least once.
   
   Now for the default behavior @dawidwys mentioned, the user has to call the 
`setMessageIdentifiers` if they are using the custom deserialiser which you 
would only use to send your custom `correlationId` with. In the default 
deserializer the method is called automatically so no need for the user to do 
anything. 
   
   I really hope that did the trick and i am more than happy to do more if 
needed :)



This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [flink] senegalo commented on pull request #12056: [FLINK-17502] [flink-connector-rabbitmq] RMQSource refactor

[GitBox]

senegalo commented on pull request #12056:
URL: https://github.com/apache/flink/pull/12056#issuecomment-650738330


   @dawidwys @austince Thanks for the "get well" wishes :) i am doing much 
better now.
   
   I finished some of the comments you guys addressed i also had some comments 
/ clarification requests on some other.
   
   can't stress enough .. thank you for your time :)



This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [flink] senegalo commented on pull request #12056: [FLINK-17502] [flink-connector-rabbitmq] RMQSource refactor

[GitBox]

senegalo commented on pull request #12056:
URL: https://github.com/apache/flink/pull/12056#issuecomment-648724664


   sorry for the no update for the past week .. it's pollen season and i was 
suffering from it a bit last weekend ... will do my best to finish that all 
during the upcoming weekend



This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [flink] senegalo commented on pull request #12056: [FLINK-17502] [flink-connector-rabbitmq] RMQSource refactor

[GitBox]

senegalo commented on pull request #12056:
URL: https://github.com/apache/flink/pull/12056#issuecomment-644549028


   awesome .. will look into the changes and hopefully be done by them next 
weekend.



This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [flink] senegalo commented on pull request #12056: [FLINK-17502] [flink-connector-rabbitmq] RMQSource refactor

[GitBox]

senegalo commented on pull request #12056:
URL: https://github.com/apache/flink/pull/12056#issuecomment-644117975


   Hey @austince  @dawidwys,
   
   Did you guys had the time to look at the changes ? i really want to close 
that PR to start working on some other stuff "limited free time managment :) 
can't start something new till i finish something old !"
   
   Anyway looking forward hearing from you.
   



This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [flink] senegalo commented on pull request #12056: [FLINK-17502] [flink-connector-rabbitmq] RMQSource refactor

[GitBox]

senegalo commented on pull request #12056:
URL: https://github.com/apache/flink/pull/12056#issuecomment-640193242


   @dawidwys @austince no worries at all .. I've been having some computer 
trouble as well lately my 7 year old laptop decided it was time ! lucky for me 
that was an excuse for a new XPS 15 !
   
   It seems like i miss understood you before so here is another iteration of 
what i got from you the above.
   
   > we should not expose the concrete internal implementation of RMQCollector 
from RMQSource. It's better to have an interface defined in 
RMQDeserializationSchema
   
   Done now we have a new interface that lives in the `RMQDeserializationSchema`
   
   >the correlation check will pass only for the first element of a 
correlationId from a batch. We should account for a situation when from a 
single RMQ record (single correlationId) we produce multiple flink record
   
   You are absolutely right !! i have no clue how this slipped by. i fixed that 
by moving all the checks into one method `setMessageIdentifiers` this one takes 
both the `correlationID` and the `deliveryTag` of the message and sets a flag 
depending on if the conditions where met or not.
   Calling `collect` without calling this method first will throw an exception.
   
   >In the end I think with @austince we discussed a slightly different 
RMQCollector interface
   
   Sorry again for the missunderstanding :)
   
   Let me know if this does the deed or not. 
   
   



This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [flink] senegalo commented on pull request #12056: [FLINK-17502] [flink-connector-rabbitmq] RMQSource refactor

[GitBox]

senegalo commented on pull request #12056:
URL: https://github.com/apache/flink/pull/12056#issuecomment-636692690


   @austince  and @dawidwys it seems there is an issue with azure can't rerun 
the pipeline and when it ran on Saturday i couldn't find mvn in the e2e tests.
   
   Please advise.



This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [flink] senegalo commented on pull request #12056: [FLINK-17502] [flink-connector-rabbitmq] RMQSource refactor

[GitBox]

senegalo commented on pull request #12056:
URL: https://github.com/apache/flink/pull/12056#issuecomment-636502616


   @flinkbot run azure



This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [flink] senegalo commented on pull request #12056: [FLINK-17502] [flink-connector-rabbitmq] RMQSource refactor

[GitBox]

senegalo commented on pull request #12056:
URL: https://github.com/apache/flink/pull/12056#issuecomment-636442005


   Hemm .. previous run of Travis failed in the e2e tests because it couldn't 
find maven ! 
   ```
   Invoking mvn with '/home/vsts/maven_cache/apache-maven-3.2.5/bin/mvn 
-Dmaven.wagon.http.pool=false --settings 
/home/vsts/work/1/s/tools/ci/google-mirror-settings.xml 
-Dorg.slf4j.simpleLogger.showDateTime=true 
-Dorg.slf4j.simpleLogger.dateTimeFormat=HH:mm:ss.SSS 
-Dorg.slf4j.simpleLogger.log.org.apache.maven.cli.transfer.Slf4jMavenTransferListener=warn
 --no-snapshot-updates -B -Dhadoop.version=2.8.3 -Dinclude_hadoop_aws 
-Dscala-2.11  clean install -Dmaven.repo.local=/home/vsts/work/1/.m2/repository 
-Dflink.convergence.phase=install -Pcheck-convergence -Dflink.forkCount=2 
-Dflink.forkCountTestPackage=2 -Dmaven.javadoc.skip=true -U -DskipTests'
   2020-05-30T13:52:11.0916471Z /home/vsts/work/1/s/tools/ci/maven-utils.sh: 
line 29: /home/vsts/maven_cache/apache-maven-3.2.5/bin/mvn: No such file or 
directory
   ```
   
   Mr. @flinkbot  run azure



This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [flink] senegalo commented on pull request #12056: [FLINK-17502] [flink-connector-rabbitmq] RMQSource refactor

[GitBox]

senegalo commented on pull request #12056:
URL: https://github.com/apache/flink/pull/12056#issuecomment-636331412


   Mr. @flinkbot run azure



This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [flink] senegalo commented on pull request #12056: [FLINK-17502] [flink-connector-rabbitmq] RMQSource refactor

[GitBox]

senegalo commented on pull request #12056:
URL: https://github.com/apache/flink/pull/12056#issuecomment-634251720


   Thanks a lot for the time and effort guys. Will review your suggestions on
   Friday and do the required changes :)
   
   On Tue 26. May 2020 at 17:24, Austin Cawley-Edwards <
   notificati...@github.com> wrote:
   
   > I like the idea of passing the correlationId in with each collected
   > record - I think it makes more sense from a users point of view than the
   > setCorrelationId semantics, and the allowing multiple uses of a
   > correlationId in a single pass sounds like something that should be
   > hidden from the user anyway. @senegalo ,
   > what do you think?
   >
   > —
   > You are receiving this because you were mentioned.
   > Reply to this email directly, view it on GitHub
   > , or
   > unsubscribe
   > 

   > .
   >
   



This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [flink] senegalo commented on pull request #12056: [FLINK-17502] [flink-connector-rabbitmq] RMQSource refactor

[GitBox]

senegalo commented on pull request #12056:
URL: https://github.com/apache/flink/pull/12056#issuecomment-631906626


   @aljoscha @austince just a ping in case you missed my last comment 



This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [flink] senegalo commented on pull request #12056: [FLINK-17502] [flink-connector-rabbitmq] RMQSource refactor

[GitBox]

senegalo commented on pull request #12056:
URL: https://github.com/apache/flink/pull/12056#issuecomment-629626940


   Please Mr @flinkbot run azure



This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [flink] senegalo commented on pull request #12056: [FLINK-17502] [flink-connector-rabbitmq] RMQSource refactor

[GitBox]

senegalo commented on pull request #12056:
URL: https://github.com/apache/flink/pull/12056#issuecomment-629626611


   Okay i pushed some new changes and let me explain my endeavors for the last 
3 hours !
   
   ### The Goal
   * combine body / correlation id parsing one go
   * conform with the changes made by #12093 
   * having a 1 to N relation between an AMQP Delivery and the parsed records 
that would be passed to the collector.
   
   ### Why i failed to implement the suggestion
   
   I did exactly what you described above: 
   * i passed the collector to the interface `RMQDeserializationSchema` 
   * When processMessage is called it would extract the record(s) and the 
correlation id then call the collector newly implemented method`collect(OUT 
records, String correlationID)` 
   * The collector would then stash the correlationId in a private var for me 
to use it in the `synchronized` block. 
   
   The problem is that "or at least to my understanding" if you call `collect` 
on the collector the data is already out of the source operator. 
   This would constitute a problem if the `autoAck` is false because we need to 
decide if we are going to add the record(s) to the collector or not based on if 
we've seen this ID before.
   Since we were doing both in one go it was impossible without a lot of 
hacking around in the code.
   
   ### Alternative solution pushed
   
   *  `RMQDeserializationSchema` would in one go deserialize both the record(s) 
and the correlation ID then return an instance of `RMQDeserializedMessage` 
which is just a wrapper class around those values.
   * the `RMQSource` would call it's `parseMessage` method that would decide 
which method to use to deserialize the message either the old way or using the 
`RMQDeserializationSchema` then return for either one of those an instance of 
`RMQDeserializedMessage`. 
   * If `autoAck` is false then the synchronized block could easily access the 
`correlationID` from the `RMQDeserializedMessage` using the `getCorrelationID` 
metohd.
   * The collector collects the record(s) from the 
`RMQDeserializedMessage#getMessages` which returns a `List`
   * Finally the `RMQCollector` now has a `collect(List records)` where i 
just iterate over the records produced by the single AMQP delivery and call the 
normal `collect(OUT record)`
   
   Hope that solution makes sense.



This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [flink] senegalo commented on pull request #12056: [FLINK-17502] [flink-connector-rabbitmq] RMQSource refactor

[GitBox]

senegalo commented on pull request #12056:
URL: https://github.com/apache/flink/pull/12056#issuecomment-628393200


   Super thanks a lot for the clarification. 
   will rebase and do the changes over the weekend as i have a bit of a cold :( 



This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [flink] senegalo commented on pull request #12056: [FLINK-17502] [flink-connector-rabbitmq] RMQSource refactor

[GitBox]

senegalo commented on pull request #12056:
URL: https://github.com/apache/flink/pull/12056#issuecomment-626194323


   Had some issues with some dependencies that was missing for random reasons 
so i wasn’t able to run the tests locally. I fixed them now. Can you please Mr 
@flinkbot run azure



This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org