Added TopicPartitionToSwimLaneMapping constructor argument eventuate-…

…foundation/eventuate-messaging-kafka#57
eventuate-foundation · Aug 6, 2022 · 2a7d109 · 2a7d109
1 parent 24ea7c3
commit 2a7d109
Showing 1 changed file with 3 additions and 2 deletions.
diff --git a/...-e2e-tests/src/test/java/io/eventuate/tram/connector/kafka/EventuateTramCdcKafkaTest.java b/...-e2e-tests/src/test/java/io/eventuate/tram/connector/kafka/EventuateTramCdcKafkaTest.java
@@ -6,8 +6,9 @@
 import io.eventuate.messaging.kafka.basic.consumer.EventuateKafkaConsumerConfigurationProperties;
 import io.eventuate.messaging.kafka.basic.consumer.KafkaConsumerFactory;
 import io.eventuate.messaging.kafka.common.EventuateKafkaConfigurationProperties;
-import io.eventuate.messaging.kafka.spring.common.EventuateKafkaPropertiesConfiguration;
 import io.eventuate.messaging.kafka.consumer.MessageConsumerKafkaImpl;
+import io.eventuate.messaging.kafka.consumer.OriginalTopicPartitionToSwimLaneMapping;
+import io.eventuate.messaging.kafka.spring.common.EventuateKafkaPropertiesConfiguration;
 import io.eventuate.messaging.kafka.spring.consumer.KafkaConsumerFactoryConfiguration;
 import io.eventuate.tram.connector.AbstractTramCdcTest;
 import org.junit.runner.RunWith;
@@ -42,7 +43,7 @@ public static class Config {
   protected void createConsumer(String topic, Consumer<String> consumer) {
     MessageConsumerKafkaImpl messageConsumerKafka = new MessageConsumerKafkaImpl(eventuateKafkaConfigurationProperties.getBootstrapServers(),
             EventuateKafkaConsumerConfigurationProperties.empty(),
-            kafkaConsumerFactory);
+            kafkaConsumerFactory, new OriginalTopicPartitionToSwimLaneMapping());
 
     messageConsumerKafka.subscribe(subscriberId,
             ImmutableSet.of(topic),