Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[FLINK-3058] Add support for Kafka 0.9.0.0 #1489

Closed

Conversation

rmetzger
Copy link
Contributor

@rmetzger rmetzger commented Jan 6, 2016

For adding Kafka 0.9.0.0 support, this commit changes the following:

  • Split up of the kafka connector into a flink-connector-kafka-(base|0.9|0.8) with different dependencies
  • The base package contains common test cases and implementations (for example the producer for 0.9 and 0.8 relies on exactly the same code)
  • the 0.8 package contains a kafka connector implementation against the SimpleConsumer (low level) API of Kafka 0.8. There are some additional tests for the ZK offset committing
  • The 0.9 package relies on the new Consumer API of Kafka 0.9.0.0
  • Support for metrics for all producers and the 0.9 consumer through Flink's accumulators.

I've tested the change on a Kafka 0.9 cluster with 2 brokers. I verified the behavior for broker failures and task manager failures using the state machine demo.

@tillrohrmann
Copy link
Contributor

Test cases Kafka09ITCase.testMultipleSourcesOnePartition and Kafka08ITCase.testOffsetInZookeeper are failing in Travis build.

@rmetzger
Copy link
Contributor Author

rmetzger commented Jan 8, 2016

There are some instabilities with the new Kafka 0.9 code. I'll look into it soon.

@nielsbasjes
Copy link
Contributor

I read that Kafka 0.9 supports Kerberos authentication (I have not yet tried this). Is that supported in this first release or should I open a Jira ticket for that?

@StephanEwen
Copy link
Contributor

We have not looked into how Kafka uses Kerberos, yet, so a ticket would be good.

@StephanEwen
Copy link
Contributor

This PR changes the name of the KafkaConsumer classes. Both of them are now called FlinkKafkaConsumer in the exact same namespace, and only differ in their Maven project.

I think that is dangerous (classes with exact same qualified name). We have seem many cases where people work with unclean dependencies, which would result in a name clash if both dependencies are accidentally included.
The end result being that the wrong class is used, the connector does not work, and is non trivial to recognize that for users.

I would vote for the following:

  • Put a qualifier either in the class name or the package: .connectors.kafka.FlinkKafkaConsumer08 or .connectors.kafka08.FlinkKafkaConsumer
  • Keep the current 0.8 classes for compatibility and deprecate them.

@rmetzger
Copy link
Contributor Author

Okay, I'll rename the consumers and producers to include the version. (FlinkKafkaConsumer08 and so on.)

@aljoscha
Copy link
Contributor

I'm not sure because I don't know everything about it but why not rename KafkaServerProvider to something that more clearly says what it is, like KafkaTestEnvironment.

@aljoscha
Copy link
Contributor

Are you sure it is a good idea to give the same name to the producers/consumers for 0.8 and 0.9? Because now we have two org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer (and Producer) and what that refers to depends on what jar files are loaded first, correct?

@rmetzger rmetzger force-pushed the flink3058-second-rebased-rebased branch 2 times, most recently from 4579d0d to 0640284 Compare January 19, 2016 11:41
@rmetzger
Copy link
Contributor Author

Thank you all for the comments. I renamed the Consumers again to include the version, I added deprecated 081 and 092 consumers.
Do you think we should also add a deprecated FlinkKafkaProducer? (The producers now have a 08 and 09 suffix as well).
I also worked on the test stability. Lets see what travis says.

If there are no further comments, I'll soon rebase and merge the pull request.

@StephanEwen
Copy link
Contributor

I would leave a deprecated producer, just to make user's life easier...

@rmetzger rmetzger force-pushed the flink3058-second-rebased-rebased branch from 0141fea to 0288c83 Compare January 20, 2016 09:40
For adding Kafka 0.9.0.0 support, this commit changes the following:
- Split up of the kafka connector into a flink-connector-kafka-(base|0.9|0.8) with different dependencies
- The base package contains common test cases, classes and implementations (the producer for 0.9 and 0.8 relies on exactly the same code)
- the 0.8 package contains a kafka connector implementation against the SimpleConsumer (low level) API of Kafka 0.8. There are some additional tests for the ZK offset committing
- The 0.9 package relies on the new Consumer API of Kafka 0.9.0.0
- Support for metrics for all producers and the 0.9 consumer through Flink's accumulators.
@rmetzger rmetzger force-pushed the flink3058-second-rebased-rebased branch from 0288c83 to 1c2b0b7 Compare January 20, 2016 15:42
@rmetzger
Copy link
Contributor Author

I addressed all concerns and rebased to master.

Once the tests have passed, I'll merge the change.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Projects
None yet
5 participants