Skip to content

Updates Kafka client from 2.8.1 to 3.3.1 #130

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

Merged
andsel merged 9 commits into from
Dec 16, 2022
Merged

Updates Kafka client from 2.8.1 to 3.3.1 #130

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
11211eb
Updates Kafka client from 2.8.1 to 3.3.1
andsel Nov 10, 2022
afe3f1c
Missed update of doc tag
andsel Nov 29, 2022
1bb9d3f
Missed update of doc tag
andsel Nov 29, 2022
10b3ac5
Bumped new major version of the plugin, pins Logstash 8.3+ and update…
andsel Dec 12, 2022
6437ed2
Fix to run bundler for snapshot 8.x (main) after PR #14667 has exclud…
andsel Dec 13, 2022
1ea9f31
Nitpick, moved comment related dependency version close to version de…
andsel Dec 13, 2022
b2568f0
Renable 'default' value for client_dns_lookup but falling back to 'us…
andsel Dec 14, 2022
f4de51c
Minor, updated the changelog statement
andsel Dec 14, 2022
6448983
Removed commented test code and provided better description
andsel Dec 16, 2022
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
2 changes: 1 addition & 1 deletion .ci/run.sh
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@ env

set -ex

export KAFKA_VERSION=2.8.1
export KAFKA_VERSION=3.3.1
./kafka_test_setup.sh

jruby -rbundler/setup -S rspec -fd
Expand Down
21 changes: 12 additions & 9 deletions build.gradle
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,9 @@ group "org.logstash.integrations"

sourceCompatibility = JavaVersion.VERSION_1_8

String confluentKafkaVersion = '7.3.0'
String apacheKafkaVersion = '3.3.1'

buildscript {
repositories {
mavenCentral()
Expand All @@ -42,31 +45,31 @@ repositories {
}

dependencies {
implementation('io.confluent:kafka-avro-serializer:6.2.2') {
implementation("io.confluent:kafka-avro-serializer:${confluentKafkaVersion}") {
exclude group: 'org.apache.kafka', module:'kafka-clients'
}
implementation('io.confluent:kafka-schema-serializer:6.2.2') {
implementation("io.confluent:kafka-schema-serializer:${confluentKafkaVersion}") {
exclude group: 'org.apache.kafka', module:'kafka-clients'
}
implementation 'io.confluent:common-config:6.2.2'
implementation "io.confluent:common-config:${confluentKafkaVersion}"
implementation 'org.apache.avro:avro:1.11.0'
implementation('io.confluent:kafka-schema-registry-client:6.2.2') {
implementation("io.confluent:kafka-schema-registry-client:${confluentKafkaVersion}") {
exclude group: 'org.apache.kafka', module:'kafka-clients'
}
implementation('org.apache.kafka:kafka_2.12:2.8.1') {
implementation("org.apache.kafka:kafka_2.12:${apacheKafkaVersion}") {
// contains kafka.utils.VerifiableProperties used by kafka-schema-registry-client, explicitly exclude
// transitive dependencies
transitive = false
}
implementation 'io.confluent:common-utils:6.2.2'
implementation "io.confluent:common-utils:${confluentKafkaVersion}"
implementation 'javax.ws.rs:javax.ws.rs-api:2.1.1'
implementation 'org.glassfish.jersey.core:jersey-common:2.33'
// given https://docs.confluent.io/current/installation/versions-interoperability.html matrix
// Confluent Platform 6.2.x is Apache Kafka 2.8.x
implementation 'org.apache.kafka:kafka-clients:2.8.1'
// Confluent Platform 7.3.x is Apache Kafka 3.3.x
Comment thread
andsel marked this conversation as resolved.
Outdated
implementation "org.apache.kafka:kafka-clients:${apacheKafkaVersion}"
implementation 'com.github.luben:zstd-jni:1.5.2-2'
implementation 'org.slf4j:slf4j-api:1.7.36'
implementation 'org.lz4:lz4-java:1.7.1'
implementation 'org.lz4:lz4-java:1.8.0'
implementation 'org.xerial.snappy:snappy-java:1.1.8.4'
}
task generateGemJarRequiresFile {
Expand Down
2 changes: 1 addition & 1 deletion docs/index.asciidoc
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
:plugin: kafka
:type: integration
:no_codec:
:kafka_client: 2.8.1
:kafka_client: 3.3.1

///////////////////////////////////////////
START - GENERATED VARIABLES, DO NOT EDIT!
Expand Down
2 changes: 1 addition & 1 deletion docs/input-kafka.asciidoc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@
:plugin: kafka
:type: input
:default_codec: plain
:kafka_client: 2.8
:kafka_client: 3.3
:kafka_client_doc: 25
Comment thread
andsel marked this conversation as resolved.
Outdated

///////////////////////////////////////////
Expand Down
2 changes: 1 addition & 1 deletion docs/output-kafka.asciidoc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@
:plugin: kafka
:type: output
:default_codec: plain
:kafka_client: 2.8
:kafka_client: 3.3
:kafka_client_doc: 25
Comment thread
andsel marked this conversation as resolved.
Outdated

///////////////////////////////////////////
Expand Down
28 changes: 14 additions & 14 deletions kafka_test_setup.sh
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@ set -ex
if [ -n "${KAFKA_VERSION+1}" ]; then
echo "KAFKA_VERSION is $KAFKA_VERSION"
else
KAFKA_VERSION=2.8.1
KAFKA_VERSION=3.3.1
fi

export _JAVA_OPTIONS="-Djava.net.preferIPv4Stack=true"
Expand Down Expand Up @@ -35,19 +35,19 @@ cp spec/fixtures/jaas.config build/confluent_platform/etc/schema-registry
cp spec/fixtures/pwd build/confluent_platform/etc/schema-registry

echo "Setting up test topics with test data"
build/kafka/bin/kafka-topics.sh --create --partitions 3 --replication-factor 1 --topic logstash_integration_topic_plain --zookeeper localhost:2181
build/kafka/bin/kafka-topics.sh --create --partitions 3 --replication-factor 1 --topic logstash_integration_topic_plain_with_headers --zookeeper localhost:2181
build/kafka/bin/kafka-topics.sh --create --partitions 3 --replication-factor 1 --topic logstash_integration_topic_plain_with_headers_badly --zookeeper localhost:2181
build/kafka/bin/kafka-topics.sh --create --partitions 3 --replication-factor 1 --topic logstash_integration_topic_snappy --zookeeper localhost:2181
build/kafka/bin/kafka-topics.sh --create --partitions 3 --replication-factor 1 --topic logstash_integration_topic_lz4 --zookeeper localhost:2181
build/kafka/bin/kafka-topics.sh --create --partitions 1 --replication-factor 1 --topic logstash_integration_topic1 --zookeeper localhost:2181
build/kafka/bin/kafka-topics.sh --create --partitions 2 --replication-factor 1 --topic logstash_integration_topic2 --zookeeper localhost:2181
build/kafka/bin/kafka-topics.sh --create --partitions 3 --replication-factor 1 --topic logstash_integration_topic3 --zookeeper localhost:2181
build/kafka/bin/kafka-topics.sh --create --partitions 1 --replication-factor 1 --topic logstash_integration_gzip_topic --zookeeper localhost:2181
build/kafka/bin/kafka-topics.sh --create --partitions 1 --replication-factor 1 --topic logstash_integration_snappy_topic --zookeeper localhost:2181
build/kafka/bin/kafka-topics.sh --create --partitions 1 --replication-factor 1 --topic logstash_integration_lz4_topic --zookeeper localhost:2181
build/kafka/bin/kafka-topics.sh --create --partitions 1 --replication-factor 1 --topic logstash_integration_zstd_topic --zookeeper localhost:2181
build/kafka/bin/kafka-topics.sh --create --partitions 3 --replication-factor 1 --topic logstash_integration_partitioner_topic --zookeeper localhost:2181
build/kafka/bin/kafka-topics.sh --create --partitions 3 --replication-factor 1 --topic logstash_integration_topic_plain --bootstrap-server localhost:9092
build/kafka/bin/kafka-topics.sh --create --partitions 3 --replication-factor 1 --topic logstash_integration_topic_plain_with_headers --bootstrap-server localhost:9092
build/kafka/bin/kafka-topics.sh --create --partitions 3 --replication-factor 1 --topic logstash_integration_topic_plain_with_headers_badly --bootstrap-server localhost:9092
build/kafka/bin/kafka-topics.sh --create --partitions 3 --replication-factor 1 --topic logstash_integration_topic_snappy --bootstrap-server localhost:9092
build/kafka/bin/kafka-topics.sh --create --partitions 3 --replication-factor 1 --topic logstash_integration_topic_lz4 --bootstrap-server localhost:9092
build/kafka/bin/kafka-topics.sh --create --partitions 1 --replication-factor 1 --topic logstash_integration_topic1 --bootstrap-server localhost:9092
build/kafka/bin/kafka-topics.sh --create --partitions 2 --replication-factor 1 --topic logstash_integration_topic2 --bootstrap-server localhost:9092
build/kafka/bin/kafka-topics.sh --create --partitions 3 --replication-factor 1 --topic logstash_integration_topic3 --bootstrap-server localhost:9092
build/kafka/bin/kafka-topics.sh --create --partitions 1 --replication-factor 1 --topic logstash_integration_gzip_topic --bootstrap-server localhost:9092
build/kafka/bin/kafka-topics.sh --create --partitions 1 --replication-factor 1 --topic logstash_integration_snappy_topic --bootstrap-server localhost:9092
build/kafka/bin/kafka-topics.sh --create --partitions 1 --replication-factor 1 --topic logstash_integration_lz4_topic --bootstrap-server localhost:9092
build/kafka/bin/kafka-topics.sh --create --partitions 1 --replication-factor 1 --topic logstash_integration_zstd_topic --bootstrap-server localhost:9092
build/kafka/bin/kafka-topics.sh --create --partitions 3 --replication-factor 1 --topic logstash_integration_partitioner_topic --bootstrap-server localhost:9092
curl -s -o build/apache_logs.txt https://s3.amazonaws.com/data.elasticsearch.org/apache_logs/apache_logs.txt
cat build/apache_logs.txt | build/kafka/bin/kafka-console-producer.sh --topic logstash_integration_topic_plain --broker-list localhost:9092
cat build/apache_logs.txt | build/kafka/bin/kafka-console-producer.sh --topic logstash_integration_topic_snappy --broker-list localhost:9092 --compression-codec snappy
Expand Down
4 changes: 2 additions & 2 deletions kafka_test_teardown.sh
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,8 +3,8 @@
set -ex

echo "Unregistering test topics"
build/kafka/bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic 'logstash_integration_.*'
build/kafka/bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic 'topic_avro.*'
build/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic 'logstash_integration_.*'
build/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic 'topic_avro.*'

echo "Stopping Kafka broker"
build/kafka/bin/kafka-server-stop.sh
Expand Down
16 changes: 8 additions & 8 deletions lib/logstash-integration-kafka_jars.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,17 +1,17 @@
# AUTOGENERATED BY THE GRADLE SCRIPT. DO NOT EDIT.

require 'jar_dependencies'
require_jar('io.confluent', 'kafka-avro-serializer', '6.2.2')
require_jar('io.confluent', 'kafka-schema-serializer', '6.2.2')
require_jar('io.confluent', 'common-config', '6.2.2')
require_jar('io.confluent', 'kafka-avro-serializer', '7.3.0')
require_jar('io.confluent', 'kafka-schema-serializer', '7.3.0')
require_jar('io.confluent', 'common-config', '7.3.0')
require_jar('org.apache.avro', 'avro', '1.11.0')
require_jar('io.confluent', 'kafka-schema-registry-client', '6.2.2')
require_jar('org.apache.kafka', 'kafka_2.12', '2.8.1')
require_jar('io.confluent', 'common-utils', '6.2.2')
require_jar('io.confluent', 'kafka-schema-registry-client', '7.3.0')
require_jar('org.apache.kafka', 'kafka_2.12', '3.3.1')
require_jar('io.confluent', 'common-utils', '7.3.0')
require_jar('javax.ws.rs', 'javax.ws.rs-api', '2.1.1')
require_jar('org.glassfish.jersey.core', 'jersey-common', '2.33')
require_jar('org.apache.kafka', 'kafka-clients', '2.8.1')
require_jar('org.apache.kafka', 'kafka-clients', '3.3.1')
require_jar('com.github.luben', 'zstd-jni', '1.5.2-2')
require_jar('org.slf4j', 'slf4j-api', '1.7.36')
require_jar('org.lz4', 'lz4-java', '1.7.1')
require_jar('org.lz4', 'lz4-java', '1.8.0')
require_jar('org.xerial.snappy', 'snappy-java', '1.1.8.4')
2 changes: 1 addition & 1 deletion lib/logstash/inputs/kafka.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -92,7 +92,7 @@ class LogStash::Inputs::Kafka < LogStash::Inputs::Base
# IP addresses for a hostname, they will all be attempted to connect to before failing the
# connection. If the value is `resolve_canonical_bootstrap_servers_only` each entry will be
# resolved and expanded into a list of canonical names.
config :client_dns_lookup, :validate => ["default", "use_all_dns_ips", "resolve_canonical_bootstrap_servers_only"], :default => "default"
config :client_dns_lookup, :validate => ["use_all_dns_ips", "resolve_canonical_bootstrap_servers_only"], :default => "use_all_dns_ips"

@robbavey robbavey Dec 13, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

What did default previously map to in Kafka?

By not allowing the setting of default, this is potentially a breaking change (even though it is somewhat unlikely that a user will explicitly set default, it's a valid setting). If we choose to remove the explicit default value we need to flag it in change logs and release notes, and figure out how we choose to bundle this with future logstashes. We would also need to update the docs for the input and output

@andsel andsel Dec 13, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

I agree with you on the breaking change. Given the explanation in previous comment do you think that maybe is better if we implicitly remap default to use_all_dns_ips?. So that if one user has explicitly set default we switch implicitly to use_all_dns_ips. In worst case we let them to connect to another node while the previous setting would have resulted in connection error.

@robbavey robbavey Dec 13, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

I think that is reasonable - we should add a note in the docs and the release notes to explicitly state the change, and they are changing the value of default to be the kafka default setting, rather than the explicit default value that Kafka previously used.

@andsel andsel Dec 14, 2022 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Applied with commit b2568f0, so no anymore breaking change if the user uses default but just log a warn message.

@andsel andsel Dec 15, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

@robbavey I've removed the breaking change on the removal of default value by admitting it, deprecating and forcing to use_all_dns_ips.
The PR is read for another review, please 🙏

# The id string to pass to the server when making requests. The purpose of this
# is to be able to track the source of requests beyond just ip/port by allowing
# a logical application name to be included.
Expand Down
2 changes: 1 addition & 1 deletion lib/logstash/outputs/kafka.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,7 +86,7 @@ class LogStash::Outputs::Kafka < LogStash::Outputs::Base
# IP addresses for a hostname, they will all be attempted to connect to before failing the
# connection. If the value is `resolve_canonical_bootstrap_servers_only` each entry will be
# resolved and expanded into a list of canonical names.
config :client_dns_lookup, :validate => ["default", "use_all_dns_ips", "resolve_canonical_bootstrap_servers_only"], :default => "default"
config :client_dns_lookup, :validate => ["use_all_dns_ips", "resolve_canonical_bootstrap_servers_only"], :default => "use_all_dns_ips"
# The id string to pass to the server when making requests.
# The purpose of this is to be able to track the source of requests beyond just
# ip/port by allowing a logical application name to be included with the request
Expand Down