From: Kai Moritz <kai@juplo.de>
Date: Mon, 4 Mar 2024 14:53:39 +0000 (+0100)
Subject: WIP:refactor: Refined channel-states, introduced `ChannelState` -- ALIGN
X-Git-Tag: rebase--2024-03-05--09-07
X-Git-Url: http://juplo.de/gitweb/?a=commitdiff_plain;h=64ede95835a496e84857c38213dbf8ea451878e0;p=demos%2Fkafka%2Fchat

WIP:refactor: Refined channel-states, introduced `ChannelState` -- ALIGN
---

diff --git a/src/test/java/de/juplo/kafka/chat/backend/KafkaConfigurationIT.java b/src/test/java/de/juplo/kafka/chat/backend/KafkaConfigurationIT.java
index 86cfa301..47ddb061 100644
--- a/src/test/java/de/juplo/kafka/chat/backend/KafkaConfigurationIT.java
+++ b/src/test/java/de/juplo/kafka/chat/backend/KafkaConfigurationIT.java
@@ -2,6 +2,7 @@ package de.juplo.kafka.chat.backend;
 
 import de.juplo.kafka.chat.backend.implementation.kafka.*;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.consumer.Consumer;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -54,9 +55,13 @@ class KafkaConfigurationIT extends AbstractConfigurationWithShardingIT
 
   @AfterAll
   static void joinConsumerTasks(
+      @Autowired Consumer dataChannelConsumer,
+      @Autowired Consumer infoChannelConsumer,
       @Autowired ChannelTaskRunner channelTaskRunner)
       throws InterruptedException
   {
+    dataChannelConsumer.wakeup();
+    infoChannelConsumer.wakeup();
     channelTaskRunner.joinChannels();
   }