[GitHub] flink issue #2509: [FLINK-4280][kafka-connector] Explicit start position con...

[kl0u]

Github user kl0u commented on the issue:

https://github.com/apache/flink/pull/2509
  
Perfect!


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink issue #2509: [FLINK-4280][kafka-connector] Explicit start position con...

[tzulitai]

Github user tzulitai commented on the issue:

https://github.com/apache/flink/pull/2509
  
Thanks! If do you happen to find inappropriate changes in 
`FlinkKafkaConsumerBaseMigrationTest`, please let me know, will be happy to 
discuss and fix it :-)

Merging this to `master` now .. 


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink issue #2509: [FLINK-4280][kafka-connector] Explicit start position con...

[kl0u]

Github user kl0u commented on the issue:

https://github.com/apache/flink/pull/2509
  
Thanks @tzulitai and @rmetzger ! Of course, feel free to proceed with this.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink issue #2509: [FLINK-4280][kafka-connector] Explicit start position con...

[tzulitai]

Github user tzulitai commented on the issue:

https://github.com/apache/flink/pull/2509
  
Thanks for letting us know @kl0u!

Yes, there are other pending PRs based on this.
I just double checked the changes in `FlinkKafkaConsumerBaseMigrationTest` 
myself, and I think that they are reasonable.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink issue #2509: [FLINK-4280][kafka-connector] Explicit start position con...

[rmetzger]

Github user rmetzger commented on the issue:

https://github.com/apache/flink/pull/2509
  
@tzulitai how does this fit your timeline. Are there PRs depending this or 
is this PR blocking your in any way?
If so, I would propose that we merge it right away.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink issue #2509: [FLINK-4280][kafka-connector] Explicit start position con...

[kl0u]

Github user kl0u commented on the issue:

https://github.com/apache/flink/pull/2509
  
Hi @tzulitai and @rmetzger . I did not have time so far to look into it. I 
hope I will be able to do it till the end of the week. Is this ok?


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink issue #2509: [FLINK-4280][kafka-connector] Explicit start position con...

[tzulitai]

Github user tzulitai commented on the issue:

https://github.com/apache/flink/pull/2509
  
Last commit addresses consolidating the deserliazer settings for 
`KafkaOffsetHandlerImpl`.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink issue #2509: [FLINK-4280][kafka-connector] Explicit start position con...

[tzulitai]

Github user tzulitai commented on the issue:

https://github.com/apache/flink/pull/2509
  
Thanks for the review @rmetzger! The final 2 commits have addressed all 
your comments.

I'll also wait for @kl0u to have a look at the changes in 
`FlinkKafkaConsumerBaseMigration` before merging this to `master`.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink issue #2509: [FLINK-4280][kafka-connector] Explicit start position con...

[rmetzger]

Github user rmetzger commented on the issue:

https://github.com/apache/flink/pull/2509
  
I've tested the change locally and with great success. So once my comments 
are addressed, the change is good to be merged


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink issue #2509: [FLINK-4280][kafka-connector] Explicit start position con...

[tzulitai]

Github user tzulitai commented on the issue:

https://github.com/apache/flink/pull/2509
  
Note about 3rd commit: fixed failing `FlinkKafkaConsumerBaseMigrationTest`s 
after the rebase.

The tests were failing due to the removal of 
`AbstractFetcher#restoreOffsets(...)` method as part of the refactoring of 
offset restoration in this PR. On the other hand, the previous implementation 
of tests in `FlinkKafkaConsumerBaseMigrationTest` were too tightly coupled with 
how the connector was implemented, i.e. it was testing how the 
`AbstractFetcher` methods are called, whether `MAX_VALUE` watermark was emitted 
(which will likely change as features are added to the connector) etc, even 
though the actual purpose of the tests was simply to test states were restored 
correctly.

The 3rd commit therefore attempts to simplify 
`FlinkKafkaConsumerBaseMigrationTest` to only test legacy state restore 
behaviour. The deleted parts, IMHO, are already covered in other tests.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink issue #2509: [FLINK-4280][kafka-connector] Explicit start position con...

[rmetzger]

Github user rmetzger commented on the issue:

https://github.com/apache/flink/pull/2509
  
Thanks a lot! I have your Kafka pull requests on my todo list. I hope I get 
to it soon. I'm really sorry.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink issue #2509: [FLINK-4280][kafka-connector] Explicit start position con...

[tzulitai]

Github user tzulitai commented on the issue:

https://github.com/apache/flink/pull/2509
  
Rebased on the "flink-connectors" change.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink issue #2509: [FLINK-4280][kafka-connector] Explicit start position con...

[tzulitai]

Github user tzulitai commented on the issue:

https://github.com/apache/flink/pull/2509
  
Hi @rmetzger, I've addressed all comments. I'll leave comments inline of 
code on parts that addresses your more bigger comments, to help with the 
second-pass review.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink issue #2509: [FLINK-4280][kafka-connector] Explicit start position con...

[tzulitai]

Github user tzulitai commented on the issue:

https://github.com/apache/flink/pull/2509
  
Thanks for the review @rmetzger :)
I'll aim to address your comments and rebase by the end of this week (will 
tag you once it's ready).


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink issue #2509: [FLINK-4280][kafka-connector] Explicit start position con...

[tzulitai]

Github user tzulitai commented on the issue:

https://github.com/apache/flink/pull/2509
  
Rebased on recent Kafka consumer changes, fixed failing Kafka 0.10 
exactly-once tests, and added integration tests 
(`testStartFromEarliestOffsets`, `testStartFromLatestOffsets`, and 
`testStartFromGroupOffsets`) for the new explicit startup modes.

However, I'm bumping into Kafka consumer config errors when running the 
`testStartFromEarliestOffsets` in versions 0.9 and 0.10. Still investigating 
the issue, currently `testStartFromEarliestOffsets` is deliberately commented 
out in 0.9 and 0.10 IT tests for some early reviews.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink issue #2509: [FLINK-4280][kafka-connector] Explicit start position con...

[tzulitai]

Github user tzulitai commented on the issue:

https://github.com/apache/flink/pull/2509
  
@rmetzger @gyfora @koeninger Rebased this on the Kafka 0.10 connector and 
some other recent changes. This is ready for review now ;) I'd like to add 
tests for this after #2580, because #2580 adds a `OffsetHandler` to the Kafka 
test environment in the IT tests, which will come in handy when writing tests 
for this PR.

I'll also open a separate PR based on this one for FLINK-3123 (set specific 
offsets for startup).


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink issue #2509: [FLINK-4280][kafka-connector] Explicit start position con...

[tzulitai]

Github user tzulitai commented on the issue:

https://github.com/apache/flink/pull/2509
  
Rebasing + adding tests for the new functions now.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink issue #2509: [FLINK-4280][kafka-connector] Explicit start position con...

[gyfora]

Github user gyfora commented on the issue:

https://github.com/apache/flink/pull/2509
  
@tzulitai makes sense ! As for for the Map you are right, the 
multiple topic case slipped my mind :)


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink issue #2509: [FLINK-4280][kafka-connector] Explicit start position con...

[rmetzger]

Github user rmetzger commented on the issue:

https://github.com/apache/flink/pull/2509
  
Thank you for working on this. I gave #2369 some love today to speed up 
things ;)


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink issue #2509: [FLINK-4280][kafka-connector] Explicit start position con...

[tzulitai]

Github user tzulitai commented on the issue:

https://github.com/apache/flink/pull/2509
  
I'll rebase this PR soon, probably will also wait for Kafka 0.10 connector 
to be merged.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink issue #2509: [FLINK-4280][kafka-connector] Explicit start position con...

[tzulitai]

Github user tzulitai commented on the issue:

https://github.com/apache/flink/pull/2509
  
Hi @gyfora,
Yes, it is absolutely possible to add that. There's actually a JIRA for 
that feature too 
([FLINK-3123](https://issues.apache.org/jira/browse/FLINK-3123)), so I'd say we 
can add that feature on top of the proposed changes here, as a separate follow 
up PR after this one?

One note though, the API for that feature would need to be able to specify 
offsets for partitions of different topics, since the Kafka consumers can 
subscribe multiple topics. So, `Map` wouldn't fit this case, 
probably would be better off having a new user-facing class as the argument to 
define the offsets.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink issue #2509: [FLINK-4280][kafka-connector] Explicit start position con...

[gyfora]

Github user gyfora commented on the issue:

https://github.com/apache/flink/pull/2509
  
Hi,

I like the proposed changes, do you think it would make sense to add the 
possibility to set specific offsets on a per partition basis?

```
kafka.setStartOffsets(Map partitionOffsets)
```

I think this is extremely useful in production use.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---