chore(docker): try a better startup process with docker-compose

Dockerfile

@@ -6,8 +6,13 @@ ARG APT_PACKAGES=""
 WORKDIR /app
 COPY docker /
 
-RUN if [ -n "${APT_PACKAGES}" ]; then apt-get update -y; apt-get install -y --no-install-recommends ${APT_PACKAGES}; apt-get clean && rm -rf /var/lib/apt/lists/* /var/tmp/*; fi && \
-    if [ -n "${KESTRA_PLUGINS}" ]; then /app/kestra plugins install ${KESTRA_PLUGINS}; fi
+RUN mkdir -p /app/plugins && \
+  apt-get update -y && \
+  apt-get install -y --no-install-recommends curl wait-for-it ${APT_PACKAGES} && \
+  apt-get upgrade -y && \
+  apt-get clean && rm -rf /var/lib/apt/lists/* /var/tmp/* && \
+
+RUN if [ -n "${KESTRA_PLUGINS}" ]; then /app/kestra plugins install ${KESTRA_PLUGINS}; fi
 
 ENTRYPOINT ["docker-entrypoint.sh"]
 

docker-compose.yml

@@ -3,8 +3,6 @@ version: "3.6"
 volumes:
   zookeeper-data:
     driver: local
-  zookeeper-log:
-    driver: local
   kafka-data:
     driver: local
   elasticsearch-data:
@@ -14,30 +12,26 @@ volumes:
 
 services:
   zookeeper:
-    image: confluentinc/cp-zookeeper:7.0.1
+    image: 'bitnami/zookeeper:latest'
     volumes:
-      - zookeeper-data:/var/lib/zookeeper/data
-      - zookeeper-log:/var/lib/zookeeper/log
+      - zookeeper-data:/bitnami/zookeeper
     environment:
+      ALLOW_ANONYMOUS_LOGIN: "yes"
       ZOOKEEPER_CLIENT_PORT: 2181
-      ZOOKEEPER_LOG4J_ROOT_LOGLEVEL: WARN
-      ZOOKEEPER_TOOLS_LOG4J_LOGLEVEL: WARN
+      ZOO_LOG_LEVEL: "WARN"
 
   kafka:
-    image: confluentinc/cp-kafka:7.0.1
+    image: 'bitnami/kafka:latest'
     volumes:
-      - kafka-data:/var/lib/kafka
+      - kafka-data:/bitnami
     environment:
-      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
-      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
-      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
-      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
-      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
-      KAFKA_CONFLUENT_SUPPORT_METRICS_ENABLE: 'false'
-      KAFKA_LOG4J_LOGGERS: "kafka=WARN,kafka.producer.async.DefaultEventHandler=WARN,kafka.controller=WARN,state.change.logger=WARN"
-      KAFKA_LOG4J_ROOT_LOGLEVEL: WARN
-      KAFKA_TOOLS_LOG4J_LOGLEVEL: WARN
-    links:
+      ALLOW_PLAINTEXT_LISTENER: "yes"
+      KAFKA_CFG_ZOOKEEPER_CONNECT: zookeeper:2181
+      KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_CFG_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+      KAFKA_CFG_TRANSACTION_STATE_LOG_MIN_ISR: 1
+      KAFKA_CFG_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
+    depends_on:
       - zookeeper
 
   elasticsearch:
@@ -59,7 +53,7 @@ services:
 
   kestra:
     image: kestra/kestra:develop-full
-    command: server standalone
+    entrypoint: /usr/bin/wait-for-it -t 60 kafka:9092 && /usr/bin/wait-for-it -t 60 elasticsearch:9200 && /app/kestra server standalone
     volumes:
       - kestra-data:/app/storage
       - /var/run/docker.sock:/var/run/docker.sock
@@ -71,6 +65,10 @@ services:
             client:
               properties:
                 bootstrap.servers: kafka:9092
+            defaults:
+              stream:
+                properties:
+                  state.dir: "/tmp/kestra/kafka-streams/"
           elasticsearch:
             client:
               http-hosts: http://elasticsearch:9200
@@ -88,6 +86,7 @@ services:
           url: http://localhost:8080/
     ports:
       - "8080:8080"
-    links:
+    depends_on:
       - kafka
       - zookeeper
+      - elasticsearch

runner-kafka/src/main/java/io/kestra/runner/kafka/services/KafkaAdminService.java

@@ -9,6 +9,7 @@
 import org.apache.kafka.clients.admin.ConfigEntry;
 import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.common.config.ConfigResource;
+import org.apache.kafka.common.errors.TimeoutException;
 import org.apache.kafka.common.errors.TopicExistsException;
 import io.kestra.core.metrics.MetricRegistry;
 import io.kestra.runner.kafka.configs.ClientConfig;
@@ -127,7 +128,7 @@ public void createIfNotExist(TopicsConfig topicConfig) {
         try {
             this.of().createTopics(Collections.singletonList(newTopic)).all().get();
             log.info("Topic '{}' created", newTopic.name());
-        } catch (ExecutionException | InterruptedException e) {
+        } catch (ExecutionException | InterruptedException | TimeoutException e) {
             if (e.getCause() instanceof TopicExistsException) {
                 try {
                     adminClient