KAFKA-17923 Remove old kafka version from e2e (#17673)

Reviewers: Chia-Ping Tsai <[email protected]>

Author: FrankYang0529

tests/docker/Dockerfile

@@ -77,15 +77,6 @@ RUN echo 'PermitUserEnvironment yes' >> /etc/ssh/sshd_config
 # Install binary test dependencies.
 # we use the same versions as in vagrant/base.sh
 ARG KAFKA_MIRROR="https://s3-us-west-2.amazonaws.com/kafka-packages"
-RUN mkdir -p "/opt/kafka-0.8.2.2" && chmod a+rw /opt/kafka-0.8.2.2 && curl -s "$KAFKA_MIRROR/kafka_2.11-0.8.2.2.tgz" | tar xz --strip-components=1 -C "/opt/kafka-0.8.2.2"
-RUN mkdir -p "/opt/kafka-0.9.0.1" && chmod a+rw /opt/kafka-0.9.0.1 && curl -s "$KAFKA_MIRROR/kafka_2.11-0.9.0.1.tgz" | tar xz --strip-components=1 -C "/opt/kafka-0.9.0.1"
-RUN mkdir -p "/opt/kafka-0.10.0.1" && chmod a+rw /opt/kafka-0.10.0.1 && curl -s "$KAFKA_MIRROR/kafka_2.11-0.10.0.1.tgz" | tar xz --strip-components=1 -C "/opt/kafka-0.10.0.1"
-RUN mkdir -p "/opt/kafka-0.10.1.1" && chmod a+rw /opt/kafka-0.10.1.1 && curl -s "$KAFKA_MIRROR/kafka_2.11-0.10.1.1.tgz" | tar xz --strip-components=1 -C "/opt/kafka-0.10.1.1"
-RUN mkdir -p "/opt/kafka-0.10.2.2" && chmod a+rw /opt/kafka-0.10.2.2 && curl -s "$KAFKA_MIRROR/kafka_2.11-0.10.2.2.tgz" | tar xz --strip-components=1 -C "/opt/kafka-0.10.2.2"
-RUN mkdir -p "/opt/kafka-0.11.0.3" && chmod a+rw /opt/kafka-0.11.0.3 && curl -s "$KAFKA_MIRROR/kafka_2.11-0.11.0.3.tgz" | tar xz --strip-components=1 -C "/opt/kafka-0.11.0.3"
-RUN mkdir -p "/opt/kafka-1.0.2" && chmod a+rw /opt/kafka-1.0.2 && curl -s "$KAFKA_MIRROR/kafka_2.11-1.0.2.tgz" | tar xz --strip-components=1 -C "/opt/kafka-1.0.2"
-RUN mkdir -p "/opt/kafka-1.1.1" && chmod a+rw /opt/kafka-1.1.1 && curl -s "$KAFKA_MIRROR/kafka_2.11-1.1.1.tgz" | tar xz --strip-components=1 -C "/opt/kafka-1.1.1"
-RUN mkdir -p "/opt/kafka-2.0.1" && chmod a+rw /opt/kafka-2.0.1 && curl -s "$KAFKA_MIRROR/kafka_2.12-2.0.1.tgz" | tar xz --strip-components=1 -C "/opt/kafka-2.0.1"
 RUN mkdir -p "/opt/kafka-2.1.1" && chmod a+rw /opt/kafka-2.1.1 && curl -s "$KAFKA_MIRROR/kafka_2.12-2.1.1.tgz" | tar xz --strip-components=1 -C "/opt/kafka-2.1.1"
 RUN mkdir -p "/opt/kafka-2.2.2" && chmod a+rw /opt/kafka-2.2.2 && curl -s "$KAFKA_MIRROR/kafka_2.12-2.2.2.tgz" | tar xz --strip-components=1 -C "/opt/kafka-2.2.2"
 RUN mkdir -p "/opt/kafka-2.3.1" && chmod a+rw /opt/kafka-2.3.1 && curl -s "$KAFKA_MIRROR/kafka_2.12-2.3.1.tgz" | tar xz --strip-components=1 -C "/opt/kafka-2.3.1"
@@ -106,13 +97,6 @@ RUN mkdir -p "/opt/kafka-3.8.1" && chmod a+rw /opt/kafka-3.8.1 && curl -s "$KAFK
 
 
 # Streams test dependencies
-RUN curl -s "$KAFKA_MIRROR/kafka-streams-0.10.0.1-test.jar" -o /opt/kafka-0.10.0.1/libs/kafka-streams-0.10.0.1-test.jar
-RUN curl -s "$KAFKA_MIRROR/kafka-streams-0.10.1.1-test.jar" -o /opt/kafka-0.10.1.1/libs/kafka-streams-0.10.1.1-test.jar
-RUN curl -s "$KAFKA_MIRROR/kafka-streams-0.10.2.2-test.jar" -o /opt/kafka-0.10.2.2/libs/kafka-streams-0.10.2.2-test.jar
-RUN curl -s "$KAFKA_MIRROR/kafka-streams-0.11.0.3-test.jar" -o /opt/kafka-0.11.0.3/libs/kafka-streams-0.11.0.3-test.jar
-RUN curl -s "$KAFKA_MIRROR/kafka-streams-1.0.2-test.jar" -o /opt/kafka-1.0.2/libs/kafka-streams-1.0.2-test.jar
-RUN curl -s "$KAFKA_MIRROR/kafka-streams-1.1.1-test.jar" -o /opt/kafka-1.1.1/libs/kafka-streams-1.1.1-test.jar
-RUN curl -s "$KAFKA_MIRROR/kafka-streams-2.0.1-test.jar" -o /opt/kafka-2.0.1/libs/kafka-streams-2.0.1-test.jar
 RUN curl -s "$KAFKA_MIRROR/kafka-streams-2.1.1-test.jar" -o /opt/kafka-2.1.1/libs/kafka-streams-2.1.1-test.jar
 RUN curl -s "$KAFKA_MIRROR/kafka-streams-2.2.2-test.jar" -o /opt/kafka-2.2.2/libs/kafka-streams-2.2.2-test.jar
 RUN curl -s "$KAFKA_MIRROR/kafka-streams-2.3.1-test.jar" -o /opt/kafka-2.3.1/libs/kafka-streams-2.3.1-test.jar

tests/kafkatest/directory_layout/kafka_path.py

@@ -16,7 +16,7 @@
 import importlib
 import os
 
-from kafkatest.version import get_version, KafkaVersion, DEV_BRANCH, LATEST_0_9, LATEST_3_5
+from kafkatest.version import get_version, KafkaVersion, DEV_BRANCH, LATEST_3_5
 
 
 """This module serves a few purposes:
@@ -55,11 +55,6 @@
     # This version of the file connectors does not contain ServiceLoader manifests
     LATEST_3_5.__str__(): {
         CONNECT_FILE_JAR: "libs/connect-file*.jar"
-    },
-    # TODO: This is only used in 0.8.2.x system tests, remove with KAFKA-14762
-    LATEST_0_9.__str__(): {
-        TOOLS_JAR_NAME: "libs/kafka-tools*.jar",
-        TOOLS_DEPENDANT_TEST_LIBS_JAR_NAME: "libs/{argparse4j,jackson}*.jar"
     }
 }
 

tests/kafkatest/sanity_checks/test_console_consumer.py

@@ -22,10 +22,8 @@
 
 from kafkatest.services.console_consumer import ConsoleConsumer
 from kafkatest.services.kafka import KafkaService, quorum
-from kafkatest.services.verifiable_producer import VerifiableProducer
 from kafkatest.services.zookeeper import ZookeeperService
 from kafkatest.utils.remote_account import line_count, file_exists
-from kafkatest.version import LATEST_0_8_2
 
 
 class ConsoleConsumerTest(Test):
@@ -77,24 +75,3 @@ def test_lifecycle(self, security_protocol, sasl_mechanism='GSSAPI', metadata_qu
         assert line_count(node, ConsoleConsumer.STDOUT_CAPTURE) == 0
 
         self.consumer.stop_node(node)
-
-    @cluster(num_nodes=4)
-    def test_version(self):
-        """Check that console consumer v0.8.2.X successfully starts and consumes messages."""
-        self.kafka.start()
-
-        num_messages = 1000
-        self.producer = VerifiableProducer(self.test_context, num_nodes=1, kafka=self.kafka, topic=self.topic,
-                                           max_messages=num_messages, throughput=1000)
-        self.producer.start()
-        self.producer.wait()
-
-        self.consumer.nodes[0].version = LATEST_0_8_2
-        self.consumer.new_consumer = False
-        self.consumer.consumer_timeout_ms = 1000
-        self.consumer.start()
-        self.consumer.wait()
-
-        num_consumed = len(self.consumer.messages_consumed[1])
-        num_produced = self.producer.num_acked
-        assert num_produced == num_consumed, "num_produced: %d, num_consumed: %d" % (num_produced, num_consumed)

tests/kafkatest/sanity_checks/test_performance_services.py

@@ -21,7 +21,7 @@
 from kafkatest.services.performance import ProducerPerformanceService, ConsumerPerformanceService, EndToEndLatencyService
 from kafkatest.services.performance import latency, compute_aggregate_throughput
 from kafkatest.services.zookeeper import ZookeeperService
-from kafkatest.version import DEV_BRANCH, LATEST_0_8_2, LATEST_0_9, LATEST_1_1, KafkaVersion
+from kafkatest.version import DEV_BRANCH, LATEST_2_1, KafkaVersion
 
 
 class PerformanceServiceTest(Test):
@@ -38,15 +38,8 @@ def setUp(self):
             self.zk.start()
 
     @cluster(num_nodes=5)
-    # We are keeping 0.8.2 here so that we don't inadvertently break support for it. Since this is just a sanity check,
-    # the overhead should be manageable.
-    @parametrize(version=str(LATEST_0_8_2), new_consumer=False)
-    @parametrize(version=str(LATEST_0_9), new_consumer=False)
-    @parametrize(version=str(LATEST_0_9))
-    @parametrize(version=str(LATEST_1_1), new_consumer=False)
-    @cluster(num_nodes=5)
-    @matrix(version=[str(DEV_BRANCH)], metadata_quorum=quorum.all)
-    def test_version(self, version=str(LATEST_0_9), new_consumer=True, metadata_quorum=quorum.zk):
+    @matrix(version=[str(LATEST_2_1), str(DEV_BRANCH)], metadata_quorum=quorum.all)
+    def test_version(self, version=str(LATEST_2_1), metadata_quorum=quorum.zk):
         """
         Sanity check out producer performance service - verify that we can run the service with a small
         number of messages. The actual stats here are pretty meaningless since the number of messages is quite small.
@@ -80,7 +73,7 @@ def test_version(self, version=str(LATEST_0_9), new_consumer=True, metadata_quor
 
         # check basic run of consumer performance service
         self.consumer_perf = ConsumerPerformanceService(
-            self.test_context, 1, self.kafka, new_consumer=new_consumer,
+            self.test_context, 1, self.kafka,
             topic=self.topic, version=version, messages=self.num_records)
         self.consumer_perf.group = "test-consumer-group"
         self.consumer_perf.run()

tests/kafkatest/sanity_checks/test_verifiable_producer.py

@@ -23,7 +23,7 @@
 from kafkatest.services.verifiable_producer import VerifiableProducer
 from kafkatest.services.zookeeper import ZookeeperService
 from kafkatest.utils import is_version
-from kafkatest.version import LATEST_0_8_2, LATEST_0_9, LATEST_0_10_0, LATEST_0_10_1, DEV_BRANCH, KafkaVersion
+from kafkatest.version import DEV_BRANCH, KafkaVersion
 
 
 class TestVerifiableProducer(Test):
@@ -45,10 +45,6 @@ def setUp(self):
             self.zk.start()
 
     @cluster(num_nodes=3)
-    @parametrize(producer_version=str(LATEST_0_8_2))
-    @parametrize(producer_version=str(LATEST_0_9))
-    @parametrize(producer_version=str(LATEST_0_10_0))
-    @parametrize(producer_version=str(LATEST_0_10_1))
     @matrix(producer_version=[str(DEV_BRANCH)], acks=["0", "1", "-1"], enable_idempotence=[False])
     @matrix(producer_version=[str(DEV_BRANCH)], acks=["-1"], enable_idempotence=[True])
     @matrix(producer_version=[str(DEV_BRANCH)], security_protocol=['PLAINTEXT', 'SSL'], metadata_quorum=quorum.all)
@@ -81,20 +77,7 @@ def test_simple_run(self, producer_version, acks=None, enable_idempotence=False,
         wait_until(lambda: self.producer.num_acked > 5, timeout_sec=15,
              err_msg="Producer failed to start in a reasonable amount of time.")
 
-        # using version.vstring (distutils.version.LooseVersion) is a tricky way of ensuring
-        # that this check works with DEV_BRANCH
-        # When running VerifiableProducer 0.8.X, both the current branch version and 0.8.X should show up because of the
-        # way verifiable producer pulls in some development directories into its classpath
-        #
-        # If the test fails here because 'ps .. | grep' couldn't find the process it means
-        # the login and grep that is_version() performs is slower than
-        # the time it takes the producer to produce its messages.
-        # Easy fix is to decrease throughput= above, the good fix is to make the producer
-        # not terminate until explicitly killed in this case.
-        if node.version <= LATEST_0_8_2:
-            assert is_version(node, [node.version.vstring, LATEST_0_9.vstring], logger=self.logger)
-        else:
-            assert is_version(node, [node.version.vstring], logger=self.logger)
+        assert is_version(node, [node.version.vstring], logger=self.logger)
 
         self.producer.wait()
         num_produced = self.producer.num_acked

tests/kafkatest/services/console_consumer.py

@@ -21,7 +21,7 @@
 
 from kafkatest.directory_layout.kafka_path import KafkaPathResolverMixin
 from kafkatest.services.monitor.jmx import JmxMixin, JmxTool
-from kafkatest.version import DEV_BRANCH, LATEST_0_8_2, LATEST_0_9, LATEST_0_10_0, V_0_10_0_0, V_0_11_0_0, V_2_0_0, LATEST_3_7
+from kafkatest.version import DEV_BRANCH, LATEST_3_7
 from kafkatest.services.kafka.util import fix_opts_for_new_jvm
 
 """
@@ -118,9 +118,6 @@ def __init__(self, context, num_nodes, kafka, topic, group_id="test-consumer-gro
 
         self.isolation_level = isolation_level
         self.enable_systest_events = enable_systest_events
-        if self.enable_systest_events:
-            # Only available in 0.10.0 and up
-            assert version >= V_0_10_0_0
 
         self.print_timestamp = print_timestamp
         self.jaas_override_variables = jaas_override_variables or {}
@@ -134,10 +131,6 @@ def prop_file(self, node):
         """Return a string which can be used to create a configuration file appropriate for the given node."""
         # Process client configuration
         prop_file = self.render('console_consumer.properties')
-        if hasattr(node, "version") and node.version <= LATEST_0_8_2:
-            # in 0.8.2.X and earlier, console consumer does not have --timeout-ms option
-            # instead, we have to pass it through the config file
-            prop_file += "\nconsumer.timeout.ms=%s\n" % str(self.consumer_timeout_ms)
 
         # Add security properties to the config. If security protocol is not specified,
         # use the default in the template properties.
@@ -176,28 +169,16 @@ def start_cmd(self, node):
               "%(console_consumer)s " \
               "--topic %(topic)s " \
               "--consumer.config %(config_file)s " % args
-
-        if self.new_consumer:
-            assert node.version.consumer_supports_bootstrap_server(), \
-                "new_consumer is only supported if version >= 0.9.0.0, version %s" % str(node.version)
-            if node.version <= LATEST_0_10_0:
-                cmd += " --new-consumer"
-            cmd += " --bootstrap-server %(broker_list)s" % args
-            if node.version >= V_0_11_0_0:
-                cmd += " --isolation-level %s" % self.isolation_level
-        else:
-            assert node.version < V_2_0_0, \
-                "new_consumer==false is only supported if version < 2.0.0, version %s" % str(node.version)
-            cmd += " --zookeeper %(zk_connect)s" % args
+        cmd += " --bootstrap-server %(broker_list)s" % args
+        cmd += " --isolation-level %s" % self.isolation_level
 
         if self.from_beginning:
             cmd += " --from-beginning"
 
         if self.consumer_timeout_ms is not None:
             # version 0.8.X and below do not support --timeout-ms option
             # This will be added in the properties file instead
-            if node.version > LATEST_0_8_2:
-                cmd += " --timeout-ms %s" % self.consumer_timeout_ms
+            cmd += " --timeout-ms %s" % self.consumer_timeout_ms
 
         if self.print_timestamp:
             cmd += " --property print.timestamp=true"
@@ -209,16 +190,12 @@ def start_cmd(self, node):
             cmd += " --property print.partition=true"
 
         # LoggingMessageFormatter was introduced after 0.9
-        if node.version > LATEST_0_9:
-            if node.version > LATEST_3_7:
-                cmd += " --formatter org.apache.kafka.tools.consumer.LoggingMessageFormatter"
-            else:
-                cmd += " --formatter kafka.tools.LoggingMessageFormatter"
+        if node.version > LATEST_3_7:
+            cmd += " --formatter org.apache.kafka.tools.consumer.LoggingMessageFormatter"
+        else:
+            cmd += " --formatter kafka.tools.LoggingMessageFormatter"
 
         if self.enable_systest_events:
-            # enable systest events is only available in 0.10.0 and later
-            # check the assertion here as well, in case node.version has been modified
-            assert node.version >= V_0_10_0_0
             cmd += " --enable-systest-events"
 
         if self.consumer_properties is not None:

tests/kafkatest/services/kafka/templates/kafka.properties

@@ -41,12 +41,7 @@ listener.security.protocol.map={{ listener_security_protocol_map }}
 {% if quorum_info.using_zk or quorum_info.has_brokers %}
 advertised.host.name={{ node.account.hostname }}
 advertised.listeners={{ advertised_listeners }}
-
-{% if node.version.supports_named_listeners() %}
 inter.broker.listener.name={{ interbroker_listener.name }}
-{% else %}
-security.inter.broker.protocol={{ interbroker_listener.security_protocol }}
-{% endif %}
 {% endif %}
 
 {% for k, v in listener_security_config.client_listener_overrides.items() %}

tests/kafkatest/services/kafka/util.py

@@ -16,12 +16,9 @@
 from collections import namedtuple
 
 from kafkatest.utils.remote_account import java_version
-from kafkatest.version import LATEST_0_8_2, LATEST_0_9, LATEST_0_10_0, LATEST_0_10_1, LATEST_0_10_2, LATEST_0_11_0, LATEST_1_0
 
 TopicPartition = namedtuple('TopicPartition', ['topic', 'partition'])
 
-new_jdk_not_supported = frozenset([str(LATEST_0_8_2), str(LATEST_0_9), str(LATEST_0_10_0), str(LATEST_0_10_1), str(LATEST_0_10_2), str(LATEST_0_11_0), str(LATEST_1_0)])
-
 def fix_opts_for_new_jvm(node):
     # Startup scripts for early versions of Kafka contains options
     # that not supported on latest versions of JVM like -XX:+PrintGCDateStamps or -XX:UseParNewGC.
@@ -31,12 +28,6 @@ def fix_opts_for_new_jvm(node):
     if java_ver <= 9:
         return ""
 
-    cmd = ""
-    # check kafka version for kafka node types
-    if hasattr(node, 'version'):
-        if node.version == LATEST_0_8_2 or node.version == LATEST_0_9 or node.version == LATEST_0_10_0 or node.version == LATEST_0_10_1 or node.version == LATEST_0_10_2 or node.version == LATEST_0_11_0 or node.version == LATEST_1_0:
-            cmd += "export KAFKA_GC_LOG_OPTS=\"-Xlog:gc*:file=kafka-gc.log:time,tags:filecount=10,filesize=102400\"; "
-            cmd += "export KAFKA_JVM_PERFORMANCE_OPTS=\"-server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent -XX:MaxInlineLevel=15 -Djava.awt.headless=true\"; "
-    return cmd
+    return ""
 
 

tests/kafkatest/services/monitor/jmx.py

@@ -19,7 +19,7 @@
 from ducktape.utils.util import wait_until
 
 from kafkatest.directory_layout.kafka_path import KafkaPathResolverMixin
-from kafkatest.version import get_version, V_0_11_0_0, V_3_4_0, DEV_BRANCH
+from kafkatest.version import get_version, V_3_4_0, DEV_BRANCH
 
 class JmxMixin(object):
     """This mixin helps existing service subclasses start JmxTool on their worker nodes and collect jmx stats.
@@ -139,10 +139,7 @@ def jmxtool_version(self, node):
         # To correctly wait for requested JMX metrics to be added we need the --wait option for JmxTool. This option was
         # not added until 0.11.0.1, so any earlier versions need to use JmxTool from a newer version.
         version = get_version(node)
-        if version <= V_0_11_0_0:
-            return DEV_BRANCH
-        else:
-            return version
+        return version
 
     def jmx_class_name(self, version):
         if version <= V_3_4_0: