[jira] [Commented] (FLINK-12104) Flink Kafka fails with Incompatible KafkaProducer version / NoSuchFieldException sequenceNumbers
[
https://issues.apache.org/jira/browse/FLINK-12104?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16811250#comment-16811250
]
Tim commented on FLINK-12104:
-
I can confirm that this fix works.
> Flink Kafka fails with Incompatible KafkaProducer version /
> NoSuchFieldException sequenceNumbers
>
>
> Key: FLINK-12104
> URL: https://issues.apache.org/jira/browse/FLINK-12104
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka
>Affects Versions: 1.7.2
>Reporter: Tim
>Priority: Major
>
> FlinkKafkaProducer (in flink-connector-kafka-0.11) tries to access a field
> named `sequenceNumbers` from the KafkaProducer's TransactionManager. You can
> find this line on the [master branch
> here|[https://github.com/apache/flink/blob/d6be68670e661091d94a3c65a2704d52fc0e827c/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/FlinkKafkaProducer.java#L197].]
>
> {code:java}
> Object transactionManager = getValue(kafkaProducer, "transactionManager");
> ...
> Object sequenceNumbers = getValue(transactionManager, "sequenceNumbers");
> {code}
>
> However, the Kafka TransactionManager no longer has a "sequenceNumbers"
> field. This was changed back on 9/14/2017 (KAFKA-5494) in an effort to
> support multiple inflight requests while still guaranteeing idempotence. See
> [commit diff
> here|[https://github.com/apache/kafka/commit/5d2422258cb975a137a42a4e08f03573c49a387e#diff-f4ef1afd8792cd2a2e9069cd7ddea630].]
> Subsequently when Flink tries to "recoverAndCommit" (see
> FlinkKafkaProducer011) it fails with a "NoSuchFieldException:
> sequenceNumbers", followed by a "Incompatible KafkaProducer version".
> Given that the KafkaProducer used is so old (this change was made almost two
> years ago) are there any plans of upgrading? Or - are there some known
> compatibility issues that prevent Flink/Kafka connector from doing so?
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-12104) Flink Kafka fails with Incompatible KafkaProducer version / NoSuchFieldException sequenceNumbers
[
https://issues.apache.org/jira/browse/FLINK-12104?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16809220#comment-16809220
]
Tim commented on FLINK-12104:
-
I am sure I am, not intentionally - but due to other libs in the dependency
chain. I will try with flink-connector-kafka_2.11
Side note: I'm guessing we use reflection (Class.getDeclaredField) to access
these because they are private in the Kafka codebase, and accessing them
through the public API was not feasible given what FlinkKafkaProducer was
trying to do? Asking because in that sense the build would have failed to
compile.
> Flink Kafka fails with Incompatible KafkaProducer version /
> NoSuchFieldException sequenceNumbers
>
>
> Key: FLINK-12104
> URL: https://issues.apache.org/jira/browse/FLINK-12104
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka
>Affects Versions: 1.7.2
>Reporter: Tim
>Priority: Major
>
> FlinkKafkaProducer (in flink-connector-kafka-0.11) tries to access a field
> named `sequenceNumbers` from the KafkaProducer's TransactionManager. You can
> find this line on the [master branch
> here|[https://github.com/apache/flink/blob/d6be68670e661091d94a3c65a2704d52fc0e827c/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/FlinkKafkaProducer.java#L197].]
>
> {code:java}
> Object transactionManager = getValue(kafkaProducer, "transactionManager");
> ...
> Object sequenceNumbers = getValue(transactionManager, "sequenceNumbers");
> {code}
>
> However, the Kafka TransactionManager no longer has a "sequenceNumbers"
> field. This was changed back on 9/14/2017 (KAFKA-5494) in an effort to
> support multiple inflight requests while still guaranteeing idempotence. See
> [commit diff
> here|[https://github.com/apache/kafka/commit/5d2422258cb975a137a42a4e08f03573c49a387e#diff-f4ef1afd8792cd2a2e9069cd7ddea630].]
> Subsequently when Flink tries to "recoverAndCommit" (see
> FlinkKafkaProducer011) it fails with a "NoSuchFieldException:
> sequenceNumbers", followed by a "Incompatible KafkaProducer version".
> Given that the KafkaProducer used is so old (this change was made almost two
> years ago) are there any plans of upgrading? Or - are there some known
> compatibility issues that prevent Flink/Kafka connector from doing so?
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (FLINK-12104) Flink Kafka fails with Incompatible KafkaProducer version / NoSuchFieldException sequenceNumbers
[
https://issues.apache.org/jira/browse/FLINK-12104?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16808986#comment-16808986
]
Elias Levy commented on FLINK-12104:
The {{flink-connector-kafka-0.11}} depends on the Kafka 0.11.0.2 client, which
does have that field. The field was only removed in 1.0.0. Are you overriding
the Kafka client dependency? If so, that is your problem.
If you want to use a newer Kafka client, use the universal Kafka connection
({{flink-connector-kafka_2.11}}), which tracks the latest version of the Kafka
client.
> Flink Kafka fails with Incompatible KafkaProducer version /
> NoSuchFieldException sequenceNumbers
>
>
> Key: FLINK-12104
> URL: https://issues.apache.org/jira/browse/FLINK-12104
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka
>Affects Versions: 1.7.2
>Reporter: Tim
>Priority: Major
>
> FlinkKafkaProducer (in flink-connector-kafka-0.11) tries to access a field
> named `sequenceNumbers` from the KafkaProducer's TransactionManager. You can
> find this line on the [master branch
> here|[https://github.com/apache/flink/blob/d6be68670e661091d94a3c65a2704d52fc0e827c/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/FlinkKafkaProducer.java#L197].]
>
> {code:java}
> Object transactionManager = getValue(kafkaProducer, "transactionManager");
> ...
> Object sequenceNumbers = getValue(transactionManager, "sequenceNumbers");
> {code}
>
> However, the Kafka TransactionManager no longer has a "sequenceNumbers"
> field. This was changed back on 9/14/2017 (KAFKA-5494) in an effort to
> support multiple inflight requests while still guaranteeing idempotence. See
> [commit diff
> here|[https://github.com/apache/kafka/commit/5d2422258cb975a137a42a4e08f03573c49a387e#diff-f4ef1afd8792cd2a2e9069cd7ddea630].]
> Subsequently when Flink tries to "recoverAndCommit" (see
> FlinkKafkaProducer011) it fails with a "NoSuchFieldException:
> sequenceNumbers", followed by a "Incompatible KafkaProducer version".
> Given that the KafkaProducer used is so old (this change was made almost two
> years ago) are there any plans of upgrading? Or - are there some known
> compatibility issues that prevent Flink/Kafka connector from doing so?
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
