Refactor

Author: cristianrgreco

package-lock.json

@@ -32,6 +32,7 @@
     "kafkajs": "^2.2.4"
   },
   "dependencies": {
+    "compare-versions": "^6.1.1",
     "testcontainers": "^11.1.0"
   }
 }

packages/modules/kafka/package.json

@@ -1,10 +1,10 @@
+import { satisfies } from "compare-versions";
 import {
   AbstractStartedContainer,
   BoundPorts,
   Content,
   GenericContainer,
   getContainerRuntimeClient,
-  ImageName,
   InspectResult,
   RandomUuid,
   StartedTestContainer,
@@ -66,8 +66,7 @@ export class KafkaContainer extends GenericContainer {
   constructor(image: string) {
     super(image);
 
-    const parsedImage = ImageName.fromString(image);
-    if (parsedImage.image === "confluentinc/cp-kafka" && parsedImage.tag.startsWith("8.")) {
+    if (satisfies(this.imageName.tag, ">=8.0.0")) {
       this.withKraft();
     }
 
@@ -149,7 +148,11 @@ export class KafkaContainer extends GenericContainer {
   }
 
   public override async start(): Promise<StartedKafkaContainer> {
-    if (this.mode === KafkaMode.KRAFT && this.saslSslConfig && this.isLessThanCP(7, 5)) {
+    if (
+      this.mode === KafkaMode.KRAFT &&
+      this.saslSslConfig &&
+      satisfies(this.imageName.tag, `<${MIN_KRAFT_SASL_VERSION}`)
+    ) {
       throw new Error(
         `Provided Confluent Platform's version ${this.imageName.tag} is not supported in Kraft mode with sasl (must be ${MIN_KRAFT_SASL_VERSION} or above)`
       );
@@ -166,15 +169,15 @@ export class KafkaContainer extends GenericContainer {
     // exporting KAFKA_ADVERTISED_LISTENERS with the container hostname
     command += `export KAFKA_ADVERTISED_LISTENERS=${advertisedListeners}\n`;
 
-    if (this.mode !== KafkaMode.KRAFT || this.isLessThanCP(7, 4)) {
+    if (this.mode !== KafkaMode.KRAFT || satisfies(this.imageName.tag, "<7.4.0")) {
       // Optimization: skip the checks
       command += "echo '' > /etc/confluent/docker/ensure \n";
     }
     if (this.mode === KafkaMode.KRAFT) {
       if (this.saslSslConfig) {
         command += this.commandKraftCreateUser(this.saslSslConfig);
       }
-      if (this.isLessThanCP(7, 4)) {
+      if (satisfies(this.imageName.tag, "<7.4.0")) {
         command += this.commandKraft();
       }
     } else if (this.mode === KafkaMode.EMBEDDED_ZOOKEEPER) {
@@ -268,26 +271,13 @@ export class KafkaContainer extends GenericContainer {
   }
 
   private verifyMinKraftVersion() {
-    if (this.isLessThanCP(7)) {
+    if (satisfies(this.imageName.tag, `<${MIN_KRAFT_VERSION}`)) {
       throw new Error(
         `Provided Confluent Platform's version ${this.imageName.tag} is not supported in Kraft mode (must be ${MIN_KRAFT_VERSION} or above)`
       );
     }
   }
 
-  private isLessThanCP(max: number, min = 0, patch = 0): boolean {
-    if (this.imageName.tag === "latest") {
-      return false;
-    }
-    const parts = this.imageName.tag.split(".");
-    return !(
-      parts.length > 2 &&
-      (Number(parts[0]) > max ||
-        (Number(parts[0]) === max &&
-          (Number(parts[1]) > min || (Number(parts[1]) === min && Number(parts[2]) >= patch))))
-    );
-  }
-
   private commandKraftCreateUser(saslOptions: SaslSslListenerOptions): string {
     return (
       "echo 'kafka-storage format --ignore-formatted " +