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.

Sign up for GitHub

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-36821] Update to Kafka 3.9.0 #138

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
Open

[FLINK-36821] Update to Kafka 3.9.0 #138

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
99 changes: 1 addition & 98 deletions flink-connector-kafka/archunit-violations/86dfd459-67a9-4b26-9b5c-0b0bbf22681a
View file
Open in desktop

Large diffs are not rendered by default.

110 changes: 1 addition & 109 deletions flink-connector-kafka/archunit-violations/c0d94764-76a0-4c50-b617-70b1754c4612
View file
Open in desktop

Large diffs are not rendered by default.

1 change: 0 additions & 1 deletion flink-connector-kafka/archunit-violations/d853eb69-8c04-4246-9a5e-4f5911286b1d
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,4 +2,3 @@ org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema.open(org.ap
org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema.serialize(java.lang.Object, org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema$KafkaSinkContext, java.lang.Long): Argument leaf type org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema$KafkaSinkContext does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer.getPartitionOffsets(java.util.Collection, org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer$PartitionOffsetsRetriever): Argument leaf type org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer$PartitionOffsetsRetriever does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.invoke(org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer$KafkaTransactionState, java.lang.Object, org.apache.flink.streaming.api.functions.sink.SinkFunction$Context): Argument leaf type org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer$KafkaTransactionState does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition.dropLeaderData(java.util.List): Argument leaf type org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartitionLeader does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
6 changes: 6 additions & 0 deletions ...ava/org/apache/flink/streaming/connectors/kafka/internals/FlinkKafkaInternalProducer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@
import org.apache.kafka.common.Node;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.errors.ProducerFencedException;
import org.apache.kafka.common.requests.FindCoordinatorRequest;
import org.slf4j.Logger;
Expand Down Expand Up @@ -156,6 +157,11 @@ public List<PartitionInfo> partitionsFor(String topic) {
return kafkaProducer.metrics();
}

@Override
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This change seems unrelated to the version change. If it is related please could you add a unit test.

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Because the Producer interface has changed between 3.4 and 3.9, this new method is required in order for the class to compile (I should have mentioned that in the PR description).

As this class is deprecated anyway and calls the wrapped KafkaProducer's client instance ID method (which is tested), is a unit test strictly required?

public Uuid clientInstanceId(Duration timeout) {
return kafkaProducer.clientInstanceId(timeout);
}

@Override
public void close() {
throw new UnsupportedOperationException(
Expand Down
6 changes: 3 additions & 3 deletions ...r-kafka/src/test/java/org/apache/flink/connector/kafka/testutils/DockerImageVersions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,9 +24,9 @@
*/
public class DockerImageVersions {

public static final String KAFKA = "confluentinc/cp-kafka:7.4.4";
public static final String KAFKA = "confluentinc/cp-kafka:7.7.2";

public static final String SCHEMA_REGISTRY = "confluentinc/cp-schema-registry:7.4.4";
public static final String SCHEMA_REGISTRY = "confluentinc/cp-schema-registry:7.7.2";

public static final String ZOOKEEPER = "zookeeper:3.4.14";
public static final String ZOOKEEPER = "zookeeper:3.8.4";
}
2 changes: 1 addition & 1 deletion flink-sql-connector-kafka/src/main/resources/META-INF/NOTICE
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,4 +6,4 @@ The Apache Software Foundation (http://www.apache.org/).

This project bundles the following dependencies under the Apache Software License 2.0. (http://www.apache.org/licenses/LICENSE-2.0.txt)

- org.apache.kafka:kafka-clients:3.4.0
- org.apache.kafka:kafka-clients:3.9.0
12 changes: 6 additions & 6 deletions pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,10 +51,10 @@ under the License.

<properties>
<flink.version>1.20.0</flink.version>
<kafka.version>3.4.0</kafka.version>
<confluent.version>7.4.4</confluent.version>
<kafka.version>3.9.0</kafka.version>
<confluent.version>7.7.1</confluent.version>

<jackson-bom.version>2.15.2</jackson-bom.version>
<jackson-bom.version>2.16.2</jackson-bom.version>
<junit4.version>4.13.2</junit4.version>
<junit5.version>5.9.1</junit5.version>
<assertj.version>3.23.1</assertj.version>
Expand All @@ -65,16 +65,16 @@ under the License.
<byte-buddy.version>1.12.10</byte-buddy.version>
<commons-cli.version>1.5.0</commons-cli.version>
<scala.binary.version>2.12</scala.binary.version>
<scala-reflect.version>2.12.7</scala-reflect.version>
<scala-library.version>2.12.7</scala-library.version>
<scala-reflect.version>2.12.19</scala-reflect.version>
<scala-library.version>2.12.19</scala-library.version>
<snappy-java.version>1.1.10.5</snappy-java.version>
<avro.version>1.11.4</avro.version>
<guava.version>32.1.2-jre</guava.version>

<japicmp.skip>false</japicmp.skip>
<japicmp.referenceVersion>1.17.0</japicmp.referenceVersion>

<slf4j.version>1.7.32</slf4j.version>
<slf4j.version>1.7.36</slf4j.version>
<log4j.version>2.17.1</log4j.version>

<flink.parent.artifactId>flink-connector-kafka-parent</flink.parent.artifactId>
Expand Down