[GitHub] tvielgouarin commented on issue #7405: [FLINK-11249] FlinkKafkaProducer011 can not be migrated to FlinkKafkaProducer
tvielgouarin commented on issue #7405: [FLINK-11249] FlinkKafkaProducer011 can not be migrated to FlinkKafkaProducer URL: https://github.com/apache/flink/pull/7405#issuecomment-458956630 hi @pnowojski , Thanks for your last reply and your help, I have created a PR with the corresponding test you mentioned: 1. migration of `FlinkKafkaProducer` from pre-made 1.7 savepoint to master without pending transactions 2. migration of `FlinkKafkaProducer011` from pre-made 1.7 savepoint to master without pending transactions 3. migration from `FlinkKafkaProducer011` master savepoint to FlinkKafkaProducer master without pending transactions You can find it here : https://github.com/apache/flink/pull/7604 This is an automated message from the Apache Git Service. To respond to the message, please log on 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 With regards, Apache Git Services
[GitHub] tvielgouarin commented on issue #7405: [FLINK-11249] FlinkKafkaProducer011 can not be migrated to FlinkKafkaProducer
tvielgouarin commented on issue #7405: [FLINK-11249] FlinkKafkaProducer011 can not be migrated to FlinkKafkaProducer URL: https://github.com/apache/flink/pull/7405#issuecomment-458604699 I've manage to make the migration test from FlinkKakfaProducer011 --> FlinkKafkaProducer work. I will make the PR for those migration test. This is an automated message from the Apache Git Service. To respond to the message, please log on 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 With regards, Apache Git Services
[GitHub] tvielgouarin commented on issue #7405: [FLINK-11249] FlinkKafkaProducer011 can not be migrated to FlinkKafkaProducer
tvielgouarin commented on issue #7405: [FLINK-11249] FlinkKafkaProducer011 can not be migrated to FlinkKafkaProducer URL: https://github.com/apache/flink/pull/7405#issuecomment-458558892 Thanks for your reply @pnowojski, Alright, I've made the changed for not running the test against the same KafkaBroker and Zookeeper. For the pending transaction I will just remove it then, and won't test it. What about the migration test `FlinkKafkaProducer011` --> `FlinkKafkaProducer`. Is it feasible with testHarness to restore a savepoint with a different operator ? Or should I just make the PR for the `FlinkKafkaProducer` migration between version test and `FlinkKafkaProducer011` migration between version test ? This is an automated message from the Apache Git Service. To respond to the message, please log on 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 With regards, Apache Git Services
[GitHub] tvielgouarin commented on issue #7405: [FLINK-11249] FlinkKafkaProducer011 can not be migrated to FlinkKafkaProducer
tvielgouarin commented on issue #7405: [FLINK-11249] FlinkKafkaProducer011 can not be migrated to FlinkKafkaProducer URL: https://github.com/apache/flink/pull/7405#issuecomment-458186468 Hi @pnowojski, I have managed to solve the recovery problem for the Kafka Broker. I have also implemented the test for the `FlinkKafkaProducer011` across different version of flink (1.7). So the result are the same for both `FlinkKakfaProducer` and `FlinkKafkaProducer011`. For both of them I'm trying 3 different tests: 1. ` testRestoreKafkaTempDirectory` check if we can recover the committed transaction --> OK 2. `testRestorePendingTransaction`check if we can recover the pending transaction --> ERROR no pending transaction ( Not sure if this behaviour abnormal or just that we can't recover the pending transaction after a versioning ) 3. `testRestoreProducer` check if we can produce committed transaction and pending transaction after the versioning --> OK I will work on the last test FlinkKafkaProducer011 --> FlinkKafkaProducer shortly. Refactoring should also be considered. However because I'm impacting on several directories : - kafka-connector-base - kafka-connector-011 - kafka-connector I'm not sure of the best ordering for now. I will think about it, but if you have suggestion feel free. As always my code can be found here : https://github.com/tvielgouarin/flink/commits/FLINK-11249 This is an automated message from the Apache Git Service. To respond to the message, please log on 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 With regards, Apache Git Services
[GitHub] tvielgouarin commented on issue #7405: [FLINK-11249] FlinkKafkaProducer011 can not be migrated to FlinkKafkaProducer
tvielgouarin commented on issue #7405: [FLINK-11249] FlinkKafkaProducer011 can not be migrated to FlinkKafkaProducer URL: https://github.com/apache/flink/pull/7405#issuecomment-457653008 Hi @pnowojski ! You were right, I didn't saved the zookeeper. I did the modification so that the zookeeper file is also saved. I also added the test for checking if we can recover the kafka server from the tmp files `testReuseKafkaTempDirectory`. How ever the test still doesn't pass. I guess there is a problem with my `KafkaMigrationTestEnvironenmentImpl` Could it be that the topic are deleted at some point of a KafkaEnvironementTest ? Here is all the code I have worked on so far. https://github.com/tvielgouarin/flink/commit/ea55cf43c28cf3a1027b28a25e806ac80861b4d8 This is an automated message from the Apache Git Service. To respond to the message, please log on 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 With regards, Apache Git Services
[GitHub] tvielgouarin commented on issue #7405: [FLINK-11249] FlinkKafkaProducer011 can not be migrated to FlinkKafkaProducer
tvielgouarin commented on issue #7405: [FLINK-11249] FlinkKafkaProducer011 can not be migrated to FlinkKafkaProducer URL: https://github.com/apache/flink/pull/7405#issuecomment-457232346 Hi @pnowojski Thanks for you last answer, So I have tried what you suggested : "it might be possible to identify the list of files that define the internal state of Kafka that we need to archive, place it in the resources alongside of savepoint and use it during KafkaTestBase initialisation" For that I have a created a new class `KafkaMigrationTestEnvironenmentImpl` (based on `KafkaTestEnvironementImpl` ) that doesn't erase the Kafka Broker tmp folder containing the state, and a `KafkaTestMigration` (based on `KafkaTestBase` ) that points to this last Impl. (Of course this implementation would required code refactoring ). Now the producer can connect but the test doesn't pass. The processed element can't be restored after the versioning. More weird the producer doesn't seem to produce anymore I don't know if: - this is can be related to your last post on Jira . https://issues.apache.org/jira/browse/FLINK-11249 - my test isn't correct - the state recovery for the Kafka Broker isn't right https://github.com/tvielgouarin/flink/blob/FLINK-11249/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerMigrationTest.java I think I have understand how to use testHarness but as always if you look at the code it would be much appreciated. This is an automated message from the Apache Git Service. To respond to the message, please log on 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 With regards, Apache Git Services
[GitHub] tvielgouarin commented on issue #7405: [FLINK-11249] FlinkKafkaProducer011 can not be migrated to FlinkKafkaProducer
tvielgouarin commented on issue #7405: [FLINK-11249] FlinkKafkaProducer011 can not be migrated to FlinkKafkaProducer URL: https://github.com/apache/flink/pull/7405#issuecomment-456391275 hi @pnowojski , @yanghua , I come from @cjolif 's team and I am currently working on the implementation of the Migration test in order to accelerate the merge of this PR. I came here with several questions in mind. I do understand that it needs to implement: - migration tests of `FlinkKafkaProducer011` between Flink versions - migration tests of universal `FlinkKafkaProducer` between Flink versions - migration test from `FlinkKafkaProducer011` to universal `FlinkKafkaProducer` However what should be the validation conditions ( producerConfig stay the same (?) ) Furthermore I've worked on a simple implementation of Migration test for `FlinkKafkaProducer`: https://github.com/tvielgouarin/flink/commit/973756430c226a0cd5011fedc1eab1345a27cc2a Most of the time the test fail with `The producer attempted to use a producer id which is not currently assigned to its transactional id`. I'm not sure, but I guess it's because there is a different KafkaServer's instance between the snapshot version and the test, and the transactional Id assignations are lost (?). If so, do you have and idea of the the best work around ? Anyway if you can have a look at what I've done a criticize it, it would be much appreciated. This is an automated message from the Apache Git Service. To respond to the message, please log on 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 With regards, Apache Git Services
