Map confluent platform -> kafka version

Makefile

@@ -25,3 +25,7 @@ lint:
 
 test:
 	$(GOTEST) ./...
+
+.PHONY: test_functional
+test_functional:
+	$(GOTEST) -tags=functional ./...

docker-compose.yml

@@ -1,7 +1,7 @@
 version: '3.7'
 services:
   zookeeper-1:
-    image: 'confluentinc/cp-zookeeper:5.5.0'
+    image: 'confluentinc/cp-zookeeper:${CONFLUENT_PLATFORM_VERSION:-5.5.0}'
     restart: always
     environment:
       ZOOKEEPER_SERVER_ID: '1'
@@ -13,7 +13,7 @@ services:
       ZOOKEEPER_SYNC_LIMIT: '5'
       ZOOKEEPER_MAX_CLIENT_CONNS: '0'
   zookeeper-2:
-    image: 'confluentinc/cp-zookeeper:5.5.0'
+    image: 'confluentinc/cp-zookeeper:${CONFLUENT_PLATFORM_VERSION:-5.5.0}'
     restart: always
     environment:
       ZOOKEEPER_SERVER_ID: '2'
@@ -25,7 +25,7 @@ services:
       ZOOKEEPER_SYNC_LIMIT: '5'
       ZOOKEEPER_MAX_CLIENT_CONNS: '0'
   zookeeper-3:
-    image: 'confluentinc/cp-zookeeper:5.5.0'
+    image: 'confluentinc/cp-zookeeper:${CONFLUENT_PLATFORM_VERSION:-5.5.0}'
     restart: always
     environment:
       ZOOKEEPER_SERVER_ID: '3'
@@ -37,7 +37,7 @@ services:
       ZOOKEEPER_SYNC_LIMIT: '5'
       ZOOKEEPER_MAX_CLIENT_CONNS: '0'
   kafka-1:
-    image: 'confluentinc/cp-kafka:5.5.0'
+    image: 'confluentinc/cp-kafka:${CONFLUENT_PLATFORM_VERSION:-5.5.0}'
     restart: always
     environment:
       KAFKA_ZOOKEEPER_CONNECT: 'zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181'
@@ -54,7 +54,7 @@ services:
       KAFKA_DELETE_TOPIC_ENABLE: 'true'
       KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'false'
   kafka-2:
-    image: 'confluentinc/cp-kafka:5.5.0'
+    image: 'confluentinc/cp-kafka:${CONFLUENT_PLATFORM_VERSION:-5.5.0}'
     restart: always
     environment:
       KAFKA_ZOOKEEPER_CONNECT: 'zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181'
@@ -71,7 +71,7 @@ services:
       KAFKA_DELETE_TOPIC_ENABLE: 'true'
       KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'false'
   kafka-3:
-    image: 'confluentinc/cp-kafka:5.5.0'
+    image: 'confluentinc/cp-kafka:${CONFLUENT_PLATFORM_VERSION:-5.5.0}'
     restart: always
     environment:
       KAFKA_ZOOKEEPER_CONNECT: 'zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181'
@@ -88,7 +88,7 @@ services:
       KAFKA_DELETE_TOPIC_ENABLE: 'true'
       KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'false'
   kafka-4:
-    image: 'confluentinc/cp-kafka:5.5.0'
+    image: 'confluentinc/cp-kafka:${CONFLUENT_PLATFORM_VERSION:-5.5.0}'
     restart: always
     environment:
       KAFKA_ZOOKEEPER_CONNECT: 'zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181'
@@ -105,7 +105,7 @@ services:
       KAFKA_DELETE_TOPIC_ENABLE: 'true'
       KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'false'
   kafka-5:
-    image: 'confluentinc/cp-kafka:5.5.0'
+    image: 'confluentinc/cp-kafka:${CONFLUENT_PLATFORM_VERSION:-5.5.0}'
     restart: always
     environment:
       KAFKA_ZOOKEEPER_CONNECT: 'zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181'

functional_test.go

@@ -107,9 +107,30 @@ func prepareDockerTestEnvironment(ctx context.Context, env *testEnvironment) err
 		return fmt.Errorf("failed to tear down existing env: %w", err)
 	}
 
+	if version, ok := os.LookupEnv("KAFKA_VERSION"); ok {
+		env.KafkaVersion = version
+	} else {
+		// We have cp-5.5.0 as the default in the docker-compose file, so that's kafka 2.5.0.
+		env.KafkaVersion = "2.5.0"
+	}
+
+	// the mapping of confluent platform docker image versions -> kafka versions can be
+	// found here: https://docs.confluent.io/current/installation/versions-interoperability.html
+	var confluentPlatformVersion string
+	switch env.KafkaVersion {
+	case "2.5.0":
+		confluentPlatformVersion = "5.5.0"
+	case "2.4.1":
+		confluentPlatformVersion = "5.4.2"
+	default:
+		return fmt.Errorf("don't know what confluent platform version to use for kafka %s", env.KafkaVersion)
+	}
+
+
 	c := exec.Command("docker-compose", "up", "-d")
 	c.Stdout = os.Stdout
 	c.Stderr = os.Stderr
+	c.Env = append(os.Environ(), fmt.Sprintf("CONFLUENT_PLATFORM_VERSION=%s", confluentPlatformVersion))
 	err := c.Run()
 	if err != nil {
 		return fmt.Errorf("failed to run docker-compose to start test enviroment: %w", err)