diff --git a/.ci/run.sh b/.ci/run.sh index d6ac5b40..814674b0 100755 --- a/.ci/run.sh +++ b/.ci/run.sh @@ -5,10 +5,10 @@ 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 -jruby -rbundler/setup -S rspec -fd --tag integration +bundle exec rspec -fd +bundle exec rspec -fd --tag integration ./kafka_test_teardown.sh diff --git a/.travis.yml b/.travis.yml index a50fc739..afb54c91 100644 --- a/.travis.yml +++ b/.travis.yml @@ -1,2 +1,9 @@ import: -- logstash-plugins/.ci:travis/travis.yml@1.x \ No newline at end of file + - logstash-plugins/.ci:travis/defaults.yml@1.x + - logstash-plugins/.ci:travis/exec.yml@1.x + +env: + jobs: + # lock on version 8.x because use of Jackson 2.13.3 available from 8.3.0 + - ELASTIC_STACK_VERSION=8.x DOCKER_ENV=dockerjdk17.env + - SNAPSHOT=true ELASTIC_STACK_VERSION=8.x DOCKER_ENV=dockerjdk17.env \ No newline at end of file diff --git a/CHANGELOG.md b/CHANGELOG.md index 2ebaeeb0..dde620f9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,8 @@ +## Unreleased + - Changed Kafka client to 3.3.1, requires Logstash >= 8.3.0. + - Deprecated `default` value for setting `client_dns_lookup` forcing to `use_all_dns_ips` when explicitly used. + [#130](https://github.com/logstash-plugins/logstash-integration-kafka/pull/130) + ## 10.12.0 - bump kafka client to 2.8.1 [#115](https://github.com/logstash-plugins/logstash-integration-kafka/pull/115) diff --git a/build.gradle b/build.gradle index b6a164bc..3d937de8 100644 --- a/build.gradle +++ b/build.gradle @@ -27,6 +27,11 @@ group "org.logstash.integrations" sourceCompatibility = JavaVersion.VERSION_1_8 +// given https://docs.confluent.io/current/installation/versions-interoperability.html matrix +// Confluent Platform 7.3.x is Apache Kafka 3.3.x +String confluentKafkaVersion = '7.3.0' +String apacheKafkaVersion = '3.3.1' + buildscript { repositories { mavenCentral() @@ -42,31 +47,29 @@ 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' + 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 { diff --git a/docs/index.asciidoc b/docs/index.asciidoc index af8dc6d6..f11fa9f7 100644 --- a/docs/index.asciidoc +++ b/docs/index.asciidoc @@ -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! diff --git a/docs/input-kafka.asciidoc b/docs/input-kafka.asciidoc index d9537ab7..18db6f0c 100644 --- a/docs/input-kafka.asciidoc +++ b/docs/input-kafka.asciidoc @@ -2,8 +2,8 @@ :plugin: kafka :type: input :default_codec: plain -:kafka_client: 2.8 -:kafka_client_doc: 25 +:kafka_client: 3.3 +:kafka_client_doc: 33 /////////////////////////////////////////// START - GENERATED VARIABLES, DO NOT EDIT! @@ -211,6 +211,12 @@ IP addresses for a hostname, they will all be attempted to connect to before fai connection. If the value is `resolve_canonical_bootstrap_servers_only` each entry will be resolved and expanded into a list of canonical names. +[NOTE] +==== +Starting from Kafka 3 `default` value for `client.dns.lookup` value has been removed. +If explicitly configured it fallbacks to `use_all_dns_ips`. +==== + [id="plugins-{type}s-{plugin}-client_id"] ===== `client_id` diff --git a/docs/output-kafka.asciidoc b/docs/output-kafka.asciidoc index efc5ac41..30950b83 100644 --- a/docs/output-kafka.asciidoc +++ b/docs/output-kafka.asciidoc @@ -2,8 +2,8 @@ :plugin: kafka :type: output :default_codec: plain -:kafka_client: 2.8 -:kafka_client_doc: 25 +:kafka_client: 3.3 +:kafka_client_doc: 33 /////////////////////////////////////////// START - GENERATED VARIABLES, DO NOT EDIT! @@ -181,6 +181,12 @@ all IP addresses returned for a hostname before failing the connection. If set to `resolve_canonical_bootstrap_servers_only`, each entry will be resolved and expanded into a list of canonical names. +[NOTE] +==== +Starting from Kafka 3 `default` value for `client.dns.lookup` value has been removed. +If explicitly configured it fallbacks to `use_all_dns_ips`. +==== + [id="plugins-{type}s-{plugin}-client_id"] ===== `client_id` diff --git a/kafka_test_setup.sh b/kafka_test_setup.sh index 25ff1451..76b79ce6 100755 --- a/kafka_test_setup.sh +++ b/kafka_test_setup.sh @@ -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" @@ -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 diff --git a/kafka_test_teardown.sh b/kafka_test_teardown.sh index 14f94c61..4b8c7507 100755 --- a/kafka_test_teardown.sh +++ b/kafka_test_teardown.sh @@ -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 diff --git a/lib/logstash-integration-kafka_jars.rb b/lib/logstash-integration-kafka_jars.rb index 2f4a83a6..c62856dd 100644 --- a/lib/logstash-integration-kafka_jars.rb +++ b/lib/logstash-integration-kafka_jars.rb @@ -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') diff --git a/lib/logstash/inputs/kafka.rb b/lib/logstash/inputs/kafka.rb index 9662667e..5962d807 100644 --- a/lib/logstash/inputs/kafka.rb +++ b/lib/logstash/inputs/kafka.rb @@ -92,7 +92,8 @@ 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" + # Starting from Kafka 3 `default` value for `client.dns.lookup` value has been removed. If explicitly configured it fallbacks to `use_all_dns_ips`. + config :client_dns_lookup, :validate => ["default", "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. @@ -257,6 +258,7 @@ def initialize(params = {}) def register @runner_threads = [] @metadata_mode = extract_metadata_level(@decorate_events) + reassign_dns_lookup @pattern ||= java.util.regex.Pattern.compile(@topics_pattern) unless @topics_pattern.nil? check_schema_registry_parameters end diff --git a/lib/logstash/outputs/kafka.rb b/lib/logstash/outputs/kafka.rb index 3acd5b73..a73b9b49 100644 --- a/lib/logstash/outputs/kafka.rb +++ b/lib/logstash/outputs/kafka.rb @@ -86,7 +86,8 @@ 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" + # Starting from Kafka 3 `default` value for `client.dns.lookup` value has been removed. If explicitly configured it fallbacks to `use_all_dns_ips`. + config :client_dns_lookup, :validate => ["default", "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 @@ -190,6 +191,7 @@ def register logger.warn("Kafka output is configured with finite retry. This instructs Logstash to LOSE DATA after a set number of send attempts fails. If you do not want to lose data if Kafka is down, then you must remove the retry setting.", :retries => @retries) end + reassign_dns_lookup @producer = create_producer if value_serializer == 'org.apache.kafka.common.serialization.StringSerializer' diff --git a/lib/logstash/plugin_mixins/kafka/common.rb b/lib/logstash/plugin_mixins/kafka/common.rb index 6b4b3fbd..79e0d4b7 100644 --- a/lib/logstash/plugin_mixins/kafka/common.rb +++ b/lib/logstash/plugin_mixins/kafka/common.rb @@ -43,5 +43,13 @@ def set_sasl_config(props) props.put("sasl.jaas.config", sasl_jaas_config) unless sasl_jaas_config.nil? end + def reassign_dns_lookup + if @client_dns_lookup == "default" + @client_dns_lookup = "use_all_dns_ips" + logger.warn("client_dns_lookup setting 'default' value is deprecated, forced to 'use_all_dns_ips', please update your configuration") + deprecation_logger.deprecated("Deprecated value `default` for `client_dns_lookup` option; use `use_all_dns_ips` instead.") + end + end + end end end end \ No newline at end of file diff --git a/logstash-integration-kafka.gemspec b/logstash-integration-kafka.gemspec index 4c1b3d9b..e016218a 100644 --- a/logstash-integration-kafka.gemspec +++ b/logstash-integration-kafka.gemspec @@ -1,6 +1,6 @@ Gem::Specification.new do |s| s.name = 'logstash-integration-kafka' - s.version = '10.12.0' + s.version = '11.0.0' s.licenses = ['Apache-2.0'] s.summary = "Integration with Kafka - input and output plugins" s.description = "This gem is a Logstash plugin required to be installed on top of the Logstash core pipeline "+ @@ -41,7 +41,7 @@ Gem::Specification.new do |s| # Gem dependencies s.add_runtime_dependency "logstash-core-plugin-api", ">= 1.60", "<= 2.99" - s.add_runtime_dependency "logstash-core", ">= 6.5.0" + s.add_runtime_dependency "logstash-core", ">= 8.3.0" s.add_runtime_dependency 'logstash-codec-json' s.add_runtime_dependency 'logstash-codec-plain' diff --git a/spec/unit/inputs/kafka_spec.rb b/spec/unit/inputs/kafka_spec.rb index 267ad8bf..94495577 100644 --- a/spec/unit/inputs/kafka_spec.rb +++ b/spec/unit/inputs/kafka_spec.rb @@ -83,6 +83,16 @@ it "should register" do expect { subject.register }.to_not raise_error end + + context "when the deprecated `default` is specified" do + let(:config) { common_config.merge('client_dns_lookup' => 'default') } + + it 'should fallback `client_dns_lookup` to `use_all_dns_ips`' do + subject.register + + expect(subject.client_dns_lookup).to eq('use_all_dns_ips') + end + end end describe '#running' do diff --git a/spec/unit/outputs/kafka_spec.rb b/spec/unit/outputs/kafka_spec.rb index 27cdcd67..1447520f 100644 --- a/spec/unit/outputs/kafka_spec.rb +++ b/spec/unit/outputs/kafka_spec.rb @@ -22,6 +22,14 @@ expect(kafka.topic_id).to eql 'test' expect(kafka.key_serializer).to eql 'org.apache.kafka.common.serialization.StringSerializer' end + + it 'should fallback `client_dns_lookup` to `use_all_dns_ips` when the deprecated `default` is specified' do + simple_kafka_config["client_dns_lookup"] = 'default' + kafka = LogStash::Outputs::Kafka.new(simple_kafka_config) + kafka.register + + expect(kafka.client_dns_lookup).to eq('use_all_dns_ips') + end end context 'when outputting messages' do