Migrate integration tests to KRaft broker

.docker/config/broker/kafka_server_jaas.conf

@@ -1,10 +1,4 @@
 KafkaServer {
-  org.apache.kafka.common.security.scram.ScramLoginModule required
-  username="admin" password="admin"
-  user_admin="admin";
-};
-
-KafkaClient {
   org.apache.kafka.common.security.scram.ScramLoginModule required
   username="admin" password="admin";
 };

build.gradle

@@ -46,9 +46,9 @@ dependencies {
     runtimeOnly("ch.qos.logback:logback-classic")
 
     testImplementation("org.mockito:mockito-core")
-    testImplementation("org.testcontainers:junit-jupiter")
-    testImplementation("org.testcontainers:testcontainers")
-    testImplementation("org.testcontainers:kafka")
+    testImplementation("org.testcontainers:junit-jupiter:1.20.0")
+    testImplementation("org.testcontainers:testcontainers:1.20.0")
+    testImplementation("org.testcontainers:kafka:1.20.0")
     testImplementation("org.mockito:mockito-junit-jupiter:5.12.0")
     testImplementation("org.junit.jupiter:junit-jupiter-params:5.10.3")
     testImplementation("io.projectreactor:reactor-test")

src/main/java/com/michelin/ns4kafka/repository/kafka/KafkaStore.java

@@ -41,25 +41,30 @@
  */
 @Slf4j
 public abstract class KafkaStore<T> {
-    private final Map<String, T> store;
-    private final AtomicBoolean initialized = new AtomicBoolean(false);
-    private final ReentrantLock offsetUpdateLock;
-    private final Condition offsetReachedThreshold;
     @Inject
     ApplicationContext applicationContext;
+
     @Inject
     AdminClient adminClient;
+
     @Inject
     KafkaStoreProperties kafkaStoreProperties;
+
     @Inject
     @Named(TaskExecutors.SCHEDULED)
     TaskScheduler taskScheduler;
+
+    @Property(name = "ns4kafka.store.kafka.init-timeout")
+    int initTimeout;
+
+    private final Map<String, T> store;
+    private final AtomicBoolean initialized = new AtomicBoolean(false);
+    private final ReentrantLock offsetUpdateLock;
+    private final Condition offsetReachedThreshold;
     String kafkaTopic;
     Producer<String, T> kafkaProducer;
     long offsetInSchemasTopic = -1;
     long lastWrittenOffset = -1;
-    @Property(name = "ns4kafka.store.kafka.init-timeout")
-    int initTimeout;
 
     KafkaStore(String kafkaTopic, Producer<String, T> kafkaProducer) {
         this.kafkaTopic = kafkaTopic;

src/main/java/com/michelin/ns4kafka/service/executor/TopicAsyncExecutor.java

@@ -98,14 +98,19 @@ public void synchronizeTopics() {
 
             Map<ConfigResource, Collection<AlterConfigOp>> updateTopics = checkTopics.stream()
                 .map(topic -> {
-                    Map<String, String> actualConf =
-                        brokerTopics.get(topic.getMetadata().getName()).getSpec().getConfigs();
-                    Map<String, String> expectedConf =
-                        topic.getSpec().getConfigs() == null ? Map.of() : topic.getSpec().getConfigs();
+                    Map<String, String> actualConf = brokerTopics.get(topic.getMetadata().getName())
+                        .getSpec()
+                        .getConfigs();
+
+                    Map<String, String> expectedConf = topic.getSpec().getConfigs() == null
+                        ? Map.of() : topic.getSpec().getConfigs();
+
                     Collection<AlterConfigOp> topicConfigChanges = computeConfigChanges(expectedConf, actualConf);
                     if (!topicConfigChanges.isEmpty()) {
-                        ConfigResource cr =
-                            new ConfigResource(ConfigResource.Type.TOPIC, topic.getMetadata().getName());
+                        ConfigResource cr = new ConfigResource(
+                            ConfigResource.Type.TOPIC,
+                            topic.getMetadata().getName()
+                        );
                         return Map.entry(cr, topicConfigChanges);
                     }
                     return null;

src/test/java/com/michelin/ns4kafka/integration/AclIntegrationTest.java

@@ -4,6 +4,7 @@
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import com.michelin.ns4kafka.integration.TopicIntegrationTest.BearerAccessRefreshToken;
+import com.michelin.ns4kafka.integration.container.KafkaIntegrationTest;
 import com.michelin.ns4kafka.model.AccessControlEntry;
 import com.michelin.ns4kafka.model.AccessControlEntry.AccessControlEntrySpec;
 import com.michelin.ns4kafka.model.AccessControlEntry.Permission;
@@ -21,7 +22,6 @@
 import com.michelin.ns4kafka.model.RoleBinding.Verb;
 import com.michelin.ns4kafka.service.executor.AccessControlEntryAsyncExecutor;
 import com.michelin.ns4kafka.validation.TopicValidator;
-import io.micronaut.context.annotation.Property;
 import io.micronaut.http.HttpMethod;
 import io.micronaut.http.HttpRequest;
 import io.micronaut.http.HttpResponse;
@@ -47,8 +47,7 @@
 import org.junit.jupiter.api.Test;
 
 @MicronautTest
-@Property(name = "micronaut.security.gitlab.enabled", value = "false")
-class AclIntegrationTest extends AbstractIntegrationTest {
+class AclIntegrationTest extends KafkaIntegrationTest {
     @Inject
     @Client("/")
     HttpClient ns4KafkaClient;

src/test/java/com/michelin/ns4kafka/integration/ApiResourcesIntegrationTest.java

@@ -3,11 +3,11 @@
 import static org.junit.jupiter.api.Assertions.assertEquals;
 
 import com.michelin.ns4kafka.controller.ApiResourcesController;
+import com.michelin.ns4kafka.integration.container.KafkaIntegrationTest;
 import com.michelin.ns4kafka.model.Metadata;
 import com.michelin.ns4kafka.model.Namespace;
 import com.michelin.ns4kafka.model.RoleBinding;
 import com.michelin.ns4kafka.validation.TopicValidator;
-import io.micronaut.context.annotation.Property;
 import io.micronaut.core.type.Argument;
 import io.micronaut.http.HttpMethod;
 import io.micronaut.http.HttpRequest;
@@ -21,8 +21,7 @@
 import org.junit.jupiter.api.Test;
 
 @MicronautTest
-@Property(name = "micronaut.security.gitlab.enabled", value = "false")
-class ApiResourcesIntegrationTest extends AbstractIntegrationTest {
+class ApiResourcesIntegrationTest extends KafkaIntegrationTest {
     @Inject
     @Client("/")
     HttpClient ns4KafkaClient;

src/test/java/com/michelin/ns4kafka/integration/ConnectIntegrationTest.java → src/test/java/com/michelin/ns4kafka/integration/ConnectorIntegrationTest.java

@@ -6,6 +6,7 @@
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import com.michelin.ns4kafka.integration.TopicIntegrationTest.BearerAccessRefreshToken;
+import com.michelin.ns4kafka.integration.container.KafkaConnectIntegrationTest;
 import com.michelin.ns4kafka.model.AccessControlEntry;
 import com.michelin.ns4kafka.model.AccessControlEntry.AccessControlEntrySpec;
 import com.michelin.ns4kafka.model.AccessControlEntry.Permission;
@@ -32,7 +33,6 @@
 import com.michelin.ns4kafka.validation.ConnectValidator;
 import com.michelin.ns4kafka.validation.TopicValidator;
 import io.micronaut.context.ApplicationContext;
-import io.micronaut.context.annotation.Property;
 import io.micronaut.http.HttpMethod;
 import io.micronaut.http.HttpRequest;
 import io.micronaut.http.HttpResponse;
@@ -53,8 +53,7 @@
 
 @Slf4j
 @MicronautTest
-@Property(name = "micronaut.security.gitlab.enabled", value = "false")
-class ConnectIntegrationTest extends AbstractIntegrationConnectTest {
+class ConnectorIntegrationTest extends KafkaConnectIntegrationTest {
     @Inject
     private ApplicationContext applicationContext;
 
@@ -75,7 +74,7 @@ class ConnectIntegrationTest extends AbstractIntegrationConnectTest {
     @BeforeAll
     void init() {
         // Create HTTP client as bean to load client configuration from application.yml
-        connectClient = applicationContext.createBean(HttpClient.class, connectContainer.getUrl());
+        connectClient = applicationContext.createBean(HttpClient.class, getConnectUrl());
 
         Namespace namespace = Namespace.builder()
             .metadata(Metadata.builder()
@@ -181,7 +180,7 @@ void shouldGetConnectClusterVersion() {
             .toBlocking()
             .retrieve(HttpRequest.GET("/"), ServerInfo.class);
 
-        assertEquals("7.4.1-ccs", actual.version());
+        assertEquals("7.7.0-ccs", actual.version());
     }
 
     @Test
@@ -470,7 +469,7 @@ void shouldRestartConnector() throws InterruptedException {
         assertEquals(HttpStatus.OK, restartResponse.status());
 
         waitForConnectorAndTasksToBeInState("ns1-co1", Connector.TaskState.RUNNING);
-        
+
         ConnectorStateInfo actual = connectClient
             .toBlocking()
             .retrieve(HttpRequest.GET("/connectors/ns1-co1/status"), ConnectorStateInfo.class);

src/test/java/com/michelin/ns4kafka/integration/ExceptionHandlerIntegrationTest.java

@@ -5,6 +5,7 @@
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import com.michelin.ns4kafka.integration.TopicIntegrationTest.BearerAccessRefreshToken;
+import com.michelin.ns4kafka.integration.container.KafkaIntegrationTest;
 import com.michelin.ns4kafka.model.AccessControlEntry;
 import com.michelin.ns4kafka.model.AccessControlEntry.AccessControlEntrySpec;
 import com.michelin.ns4kafka.model.AccessControlEntry.Permission;
@@ -23,7 +24,6 @@
 import com.michelin.ns4kafka.model.Topic;
 import com.michelin.ns4kafka.model.Topic.TopicSpec;
 import com.michelin.ns4kafka.validation.TopicValidator;
-import io.micronaut.context.annotation.Property;
 import io.micronaut.http.HttpMethod;
 import io.micronaut.http.HttpRequest;
 import io.micronaut.http.HttpResponse;
@@ -41,8 +41,7 @@
 import org.junit.jupiter.api.Test;
 
 @MicronautTest
-@Property(name = "micronaut.security.gitlab.enabled", value = "false")
-class ExceptionHandlerIntegrationTest extends AbstractIntegrationTest {
+class ExceptionHandlerIntegrationTest extends KafkaIntegrationTest {
     @Inject
     @Client("/")
     HttpClient ns4KafkaClient;