Merge branch 'RED-10106' into 'main'

RED-10106 a cleaner implementation for tenant listeners/exchanges and fixed...

See merge request fforesight/tenant-commons!20

.gitignore

@@ -38,4 +38,5 @@ gradlew.bat
 gradle/
 
 **/.gradle
-**/build
+**/build
+.DS_Store

build.gradle.kts

@@ -1,7 +1,6 @@
 plugins {
     `java-library`
     `maven-publish`
-    `kotlin-dsl`
     pmd
     checkstyle
     jacoco
@@ -13,7 +12,8 @@ val storageCommonsVersion = "2.43.0"
 val springBootVersion = "3.1.5"
 val springCloudVersion = "4.0.4"
 val springRabbitTest = "3.0.9"
-val lombokVersion = "1.18.30"
+val testContainersVersion = "1.20.1"
+
 
 dependencies {
     api("com.iqser.red.commons:storage-commons:${storageCommonsVersion}")
@@ -21,10 +21,10 @@ dependencies {
     api("org.springframework.boot:spring-boot-starter-web:${springBootVersion}")
     api("org.springframework.cloud:spring-cloud-starter-openfeign:${springCloudVersion}")
     api("org.springframework.boot:spring-boot-configuration-processor:${springBootVersion}")
-    api("org.projectlombok:lombok:${lombokVersion}")
+    testImplementation("org.springframework.boot:spring-boot-configuration-processor:${springBootVersion}")
-    runtimeOnly("org.springframework.boot:spring-boot-devtools:${springBootVersion}")
     testImplementation("org.springframework.boot:spring-boot-starter-test:${springBootVersion}")
     testImplementation("org.springframework.amqp:spring-rabbit-test:${springRabbitTest}")
+    testImplementation("org.testcontainers:rabbitmq:${testContainersVersion}")
 }
 
 group = "com.knecon.fforesight"
@@ -87,7 +87,7 @@ tasks.named<Test>("test") {
 
 sonarqube {
     properties {
-        property("sonar.login", providers.gradleProperty("sonarToken").getOrNull())
+        providers.gradleProperty("sonarToken").getOrNull()?.let { property("sonar.login", it) }
         property("sonar.host.url", "https://sonarqube.knecon.com")
     }
 }
@@ -107,4 +107,4 @@ tasks.jacocoTestReport {
 
 java {
     withJavadocJar()
-}
+}

src/main/java/com/knecon/fforesight/tenantcommons/MultiTenancyMessagingConfiguration.java

@@ -1,7 +1,7 @@
 package com.knecon.fforesight.tenantcommons;
 
-import com.fasterxml.jackson.databind.DeserializationFeature;
+import static com.knecon.fforesight.tenantcommons.ForwardTenantInterceptor.TENANT_HEADER_NAME;
-import com.fasterxml.jackson.databind.ObjectMapper;
+
 import org.springframework.amqp.core.MessagePostProcessor;
 import org.springframework.amqp.rabbit.config.ContainerCustomizer;
 import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
@@ -19,7 +19,8 @@ import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Primary;
 
-import static com.knecon.fforesight.tenantcommons.ForwardTenantInterceptor.TENANT_HEADER_NAME;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
 
 @Configuration
 @ConditionalOnClass(RabbitTemplate.class)
@@ -37,6 +38,7 @@ public class MultiTenancyMessagingConfiguration {
     @Primary
     public ContainerCustomizer<SimpleMessageListenerContainer> simpleMessageListenerContainerCustomizer(
             ObjectProvider<SimpleMessageListenerContainerCustomizer> customizers) {
+
         return container -> customizers.orderedStream().forEach((customizer) -> customizer.customize(container));
     }
 
@@ -45,39 +47,50 @@ public class MultiTenancyMessagingConfiguration {
     @Primary
     public RabbitTemplateCustomizer rabbitTemplateMultiCustomizer(
             ObjectProvider<RabbitTemplateMultiCustomizer> customizers) {
+
         return rabbitTemplate -> customizers.orderedStream().forEach((customizer) -> customizer.customize(rabbitTemplate));
     }
 
+
     @Bean
     public MessageConverter producerJackson2MessageConverter() {
+
         ObjectMapper mapper = new ObjectMapper().findAndRegisterModules();
         mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
 
         return new Jackson2JsonMessageConverter(mapper);
     }
 
+
     @Bean
     public RabbitTemplateMultiCustomizer rabbitTemplatePublishTenantIdHeaderCustomizer(
             @Qualifier("tenantIdSetterPostProcessor") MessagePostProcessor messagePostProcessor) {
+
         return template -> template.addBeforePublishPostProcessors(messagePostProcessor);
     }
 
+
     @Bean
     public SimpleMessageListenerContainerCustomizer rabbitInterceptTenantIdHeaderCustomizer(
             @Qualifier("tenantIdGetterPostProcessor") MessagePostProcessor messagePostProcessor) {
+
         return container -> container.addAfterReceivePostProcessors(messagePostProcessor);
     }
 
+
     @Bean
     public MessagePostProcessor tenantIdSetterPostProcessor() {
+
         return message -> {
             message.getMessageProperties().setHeader(TENANT_HEADER_NAME, TenantContext.getTenantId());
             return message;
         };
     }
 
+
     @Bean
     public MessagePostProcessor tenantIdGetterPostProcessor() {
+
         return message -> {
             String tenant = message.getMessageProperties().getHeader(TENANT_HEADER_NAME);
 
@@ -90,9 +103,11 @@ public class MultiTenancyMessagingConfiguration {
         };
     }
 
+
     @Bean
     public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
 
         return new RabbitAdmin(connectionFactory);
     }
+
 }

src/main/java/com/knecon/fforesight/tenantcommons/RabbitConfiguration.java

@@ -0,0 +1,44 @@
+package com.knecon.fforesight.tenantcommons;
+
+import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.config.BeanPostProcessor;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Configuration
+@Slf4j
+public class RabbitConfiguration {
+
+    @Bean
+    public BeanPostProcessor rabbitListenerContainerFactoryPostProcessor() {
+
+        return new CustomBeanPostProcessor();
+    }
+
+
+    public static class CustomBeanPostProcessor implements BeanPostProcessor {
+
+        @Override
+        public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
+
+            return bean;
+        }
+
+
+        @Override
+        public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
+
+            if (bean instanceof SimpleRabbitListenerContainerFactory factory) {
+                factory.setErrorHandler(t -> {
+                    log.error("Error occurred in Rabbit listener: ", t);
+                });
+            }
+            return bean;
+        }
+
+    }
+
+}

src/main/java/com/knecon/fforesight/tenantcommons/SimpleMessageListenerContainerCustomizer.java

@@ -6,4 +6,5 @@ import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
 public interface SimpleMessageListenerContainerCustomizer {
 
     void customize(SimpleMessageListenerContainer simpleMessageListenerContainer);
+
 }

src/main/java/com/knecon/fforesight/tenantcommons/StorageConnectionProviderImpl.java

@@ -4,7 +4,6 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.stereotype.Service;
 
 import com.iqser.red.storage.commons.service.StorageConnectionProvider;
-
 import lombok.RequiredArgsConstructor;
 
 @Service

src/main/java/com/knecon/fforesight/tenantcommons/TenantAwareTaskDecorator.java

@@ -1,8 +1,9 @@
 package com.knecon.fforesight.tenantcommons;
 
-import com.knecon.fforesight.tenantcommons.task.KneconTaskDecorator;
 import org.springframework.lang.NonNull;
 
+import com.knecon.fforesight.tenantcommons.task.KneconTaskDecorator;
+
 public class TenantAwareTaskDecorator implements KneconTaskDecorator {
 
     @Override

src/main/java/com/knecon/fforesight/tenantcommons/TenantsClient.java

@@ -2,7 +2,6 @@ package com.knecon.fforesight.tenantcommons;
 
 import java.util.List;
 
-import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.cloud.openfeign.FeignClient;
 import org.springframework.http.HttpStatus;
@@ -16,7 +15,7 @@ import org.springframework.web.bind.annotation.ResponseStatus;
 import com.knecon.fforesight.tenantcommons.model.TenantResponse;
 import com.knecon.fforesight.tenantcommons.model.UpdateDetailsRequest;
 
-@FeignClient(name = "TenantsResource", url = "${tenant-user-management-service.url}")
+@FeignClient(name = "TenantsResource", url = "${tenant-user-management-service.url}", configuration = TenantsClientConfiguration.class)
 @ResponseStatus(value = HttpStatus.OK)
 @ConditionalOnProperty(prefix = "fforesight.tenants", value = "remote", havingValue = "true")
 public interface TenantsClient extends TenantProvider {

src/main/java/com/knecon/fforesight/tenantcommons/TenantsClientConfiguration.java

@@ -0,0 +1,94 @@
+package com.knecon.fforesight.tenantcommons;
+
+import java.util.concurrent.TimeUnit;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import feign.RetryableException;
+import feign.Retryer;
+import lombok.extern.slf4j.Slf4j;
+
+@Configuration
+public class TenantsClientConfiguration {
+
+    public static final int INITIAL_BACKOFF_SECONDS = 15;
+    public static final int MAX_BACKOFF_MINUTES = 8;
+    public static final int RETRY_ATTEMPTS = 4;
+
+
+    @Bean
+    public Retryer feignRetryer() {
+
+        return new ExponentialRetryer(TimeUnit.SECONDS.toMillis(INITIAL_BACKOFF_SECONDS), TimeUnit.MINUTES.toMillis(MAX_BACKOFF_MINUTES), RETRY_ATTEMPTS);
+    }
+
+
+    @Slf4j
+    @SuppressWarnings("PMD.ProperCloneImplementation")
+    public static class ExponentialRetryer extends Retryer.Default {
+
+        private final long initialPeriod;
+        private final long maxPeriod;
+        private final int maxAttempts;
+        private int attempt = 1;
+        private long currentPeriod;
+
+
+        public ExponentialRetryer(long initialPeriod, long maxPeriod, int maxAttempts) {
+
+            this.initialPeriod = initialPeriod;
+            this.maxPeriod = maxPeriod;
+            this.maxAttempts = maxAttempts;
+            this.currentPeriod = initialPeriod;
+        }
+
+
+        @Override
+        public void continueOrPropagate(RetryableException e) {
+
+            if (attempt > maxAttempts) {
+                log.error("Max retry attempts reached. Giving up after {} attempts.", maxAttempts);
+                throw e;
+            }
+
+            log.info("Attempt {} - Retrying due to exception: {}", attempt, e.getMessage());
+
+            long interval;
+
+            if (e.retryAfter() != null) {
+                interval = e.retryAfter().getTime() - System.currentTimeMillis();
+
+                if (interval > maxPeriod) {
+                    interval = maxPeriod;
+                }
+
+                if (interval < 0L) {
+                    return;
+                }
+            } else {
+                interval = currentPeriod;
+                currentPeriod = Math.min(currentPeriod * 2, maxPeriod);
+            }
+
+            try {
+                Thread.sleep(interval);
+            } catch (InterruptedException ignored) {
+                Thread.currentThread().interrupt();
+                throw e; // Rethrow the exception if interrupted
+            }
+
+            attempt++;
+        }
+
+
+        @Override
+        public ExponentialRetryer clone() {
+
+            return new ExponentialRetryer(initialPeriod, maxPeriod, maxAttempts);
+        }
+
+    }
+
+}
+

src/main/java/com/knecon/fforesight/tenantcommons/listener/ITenantEventHandler.java

@@ -0,0 +1,12 @@
+package com.knecon.fforesight.tenantcommons.listener;
+
+import com.knecon.fforesight.tenantcommons.model.ITenantEvent;
+
+public interface ITenantEventHandler<T extends ITenantEvent> {
+
+    void handle(T event);
+
+
+    Class<T> getEventClass();
+
+}

src/main/java/com/knecon/fforesight/tenantcommons/listener/impl/TenantEventQueueCreationHandler.java

@@ -0,0 +1,53 @@
+package com.knecon.fforesight.tenantcommons.listener.impl;
+
+import java.util.Optional;
+
+import org.springframework.stereotype.Service;
+
+import com.knecon.fforesight.tenantcommons.listener.ITenantEventHandler;
+import com.knecon.fforesight.tenantcommons.model.TenantCreatedEvent;
+import com.knecon.fforesight.tenantcommons.model.TenantQueueProvider;
+import com.knecon.fforesight.tenantcommons.queue.RabbitQueueFromExchangeServiceImpl;
+import jakarta.annotation.PostConstruct;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class TenantEventQueueCreationHandler implements ITenantEventHandler<TenantCreatedEvent> {
+
+    private final RabbitQueueFromExchangeServiceImpl rabbitQueueService;
+    private final Optional<TenantQueueProvider> tenantQueueProvider;
+
+
+    @PostConstruct
+    public void init() {
+
+        log.info("TenantEventQueueCreationHandler initialized");
+    }
+
+
+    @Override
+    public void handle(TenantCreatedEvent tenantCreatedEvent) {
+
+        tenantQueueProvider.ifPresentOrElse(t -> {
+            log.info("Creating queues for new tenant {}", tenantCreatedEvent.getTenantId());
+
+            t.getTenantQueueConfigurations().forEach(tqc -> rabbitQueueService.addNewQueue(tqc.getListenerId(),
+                    tqc.getQueuePrefix(),
+                    tqc.getExchangeName(),
+                    tenantCreatedEvent.getTenantId(),
+                    tqc.getDlqName(),
+                    tqc.getArguments()));
+        }, () -> log.warn("This service does not provide any queues to create"));
+    }
+
+
+    @Override
+    public Class<TenantCreatedEvent> getEventClass() {
+
+        return TenantCreatedEvent.class;
+    }
+
+}

src/main/java/com/knecon/fforesight/tenantcommons/listener/impl/TenantEventQueueDeletionHandler.java

@@ -0,0 +1,53 @@
+package com.knecon.fforesight.tenantcommons.listener.impl;
+
+import java.util.Optional;
+
+import org.springframework.stereotype.Service;
+
+import com.knecon.fforesight.tenantcommons.listener.ITenantEventHandler;
+import com.knecon.fforesight.tenantcommons.model.TenantQueueProvider;
+import com.knecon.fforesight.tenantcommons.model.TenantResponse;
+import com.knecon.fforesight.tenantcommons.queue.RabbitQueueFromExchangeServiceImpl;
+import jakarta.annotation.PostConstruct;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class TenantEventQueueDeletionHandler implements ITenantEventHandler<TenantResponse> {
+
+    private final RabbitQueueFromExchangeServiceImpl rabbitQueueService;
+    private final Optional<TenantQueueProvider> tenantQueueProvider;
+
+
+    @PostConstruct
+    public void init() {
+
+        log.info("TenantEventQueueDeletionHandler initialized");
+    }
+
+
+    @Override
+    public void handle(TenantResponse tenantResponse) {
+
+        tenantQueueProvider.ifPresentOrElse(t -> {
+            log.info("Deleting queues for tenant {}", tenantResponse.getTenantId());
+
+            t.getTenantQueueConfigurations().forEach(tqc -> rabbitQueueService.deleteQueue(tqc.getListenerId(),
+                    tqc.getQueuePrefix(),
+                    tqc.getExchangeName(),
+                    tenantResponse.getTenantId(),
+                    tqc.getDlqName(),
+                    tqc.getArguments()));
+        }, () -> log.warn("This service does not provide any queues to delete"));
+    }
+
+
+    @Override
+    public Class<TenantResponse> getEventClass() {
+
+        return TenantResponse.class;
+    }
+
+}

src/main/java/com/knecon/fforesight/tenantcommons/model/AuthDetails.java

@@ -4,7 +4,6 @@ import java.util.HashSet;
 import java.util.Set;
 
 import com.fasterxml.jackson.annotation.JsonAlias;
-
 import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.NoArgsConstructor;

src/main/java/com/knecon/fforesight/tenantcommons/model/ITenantEvent.java

@@ -0,0 +1,7 @@
+package com.knecon.fforesight.tenantcommons.model;
+
+public interface ITenantEvent {
+
+    String getTenantId();
+
+}

src/main/java/com/knecon/fforesight/tenantcommons/model/TenantCreatedEvent.java

@@ -7,7 +7,7 @@ import lombok.NoArgsConstructor;
 @Data
 @AllArgsConstructor
 @NoArgsConstructor
-public class TenantCreatedEvent {
+public class TenantCreatedEvent implements ITenantEvent {
 
     private String tenantId;
 

src/main/java/com/knecon/fforesight/tenantcommons/model/TenantQueueProvider.java

@@ -0,0 +1,14 @@
+package com.knecon.fforesight.tenantcommons.model;
+
+import java.util.Set;
+
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+@RequiredArgsConstructor
+@Getter
+public class TenantQueueProvider {
+
+    public final Set<TenantQueueConfiguration> tenantQueueConfigurations;
+
+}

src/main/java/com/knecon/fforesight/tenantcommons/model/TenantResponse.java

@@ -12,7 +12,7 @@ import lombok.NoArgsConstructor;
 @Builder
 @AllArgsConstructor
 @NoArgsConstructor
-public class TenantResponse {
+public class TenantResponse implements ITenantEvent {
 
     private String tenantId;
     private String displayName;

src/main/java/com/knecon/fforesight/tenantcommons/model/TenantSyncEvent.java

@@ -0,0 +1,14 @@
+package com.knecon.fforesight.tenantcommons.model;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+public class TenantSyncEvent implements ITenantEvent {
+
+    private String tenantId;
+
+}

src/main/java/com/knecon/fforesight/tenantcommons/queue/RabbitQueueFromExchangeService.java

@@ -6,6 +6,7 @@ public interface RabbitQueueFromExchangeService {
 
     void addNewQueue(String listenerId, String queueNamePrefix, String exchangeName, String routingKey, String dlq, Map<String, Object> arguments);
 
+
     void deleteQueue(String listenerId, String queueNamePrefix, String exchangeName, String routingKey, String dlq, Map<String, Object> arguments);
 
 

src/main/java/com/knecon/fforesight/tenantcommons/queue/RabbitQueueFromExchangeServiceImpl.java

@@ -11,6 +11,7 @@ import org.springframework.stereotype.Service;
 
 import lombok.AccessLevel;
 import lombok.RequiredArgsConstructor;
+import lombok.SneakyThrows;
 import lombok.experimental.FieldDefaults;
 import lombok.extern.slf4j.Slf4j;
 
@@ -25,57 +26,78 @@ public class RabbitQueueFromExchangeServiceImpl implements RabbitQueueFromExchan
 
 
     @Override
+    @SneakyThrows
     public void addNewQueue(String listenerId, String queueNamePrefix, String exchangeName, String routingKey, String dlq, Map<String, Object> arguments) {
 
         String queueName = queueNamePrefix + "_" + routingKey;
         Queue queue = new Queue(queueName, true, false, false);
-        Binding binding = getBinding(exchangeName, routingKey, dlq, queueName, arguments);
+        addArgumentsToQueue(queue, dlq, arguments);
-        rabbitAdmin.declareQueue(queue);
+        Binding binding = getBinding(exchangeName, routingKey, queueName);
+
+        String returnedQueueName = rabbitAdmin.declareQueue(queue);
+        log.debug("Declared queue {}", returnedQueueName);
         rabbitAdmin.declareBinding(binding);
+        log.debug("Declared binding");
         addQueueToListener(listenerId, queueName);
     }
 
 
+    private void addArgumentsToQueue(Queue queue, String dlq, Map<String, Object> arguments) {
+
+        if (dlq != null && !dlq.isBlank()) {
+            queue.addArgument("x-dead-letter-exchange", "");
+            queue.addArgument("x-dead-letter-routing-key", dlq);
+            arguments.forEach(queue::addArgument);
+        }
+    }
+
+
+    private static Binding getBinding(String exchangeName, String routingKey, String queueName) {
+
+        return new Binding(queueName, Binding.DestinationType.QUEUE, exchangeName, routingKey, null);
+    }
+
+
+    private AbstractMessageListenerContainer getMessageListenerContainerById(String listenerId) {
+
+        log.debug("Getting message listener container by id {}", listenerId);
+        return ((AbstractMessageListenerContainer) rabbitListenerEndpointRegistry.getListenerContainer(listenerId));
+    }
+
+
     @Override
     public void deleteQueue(String listenerId, String queueNamePrefix, String exchangeName, String routingKey, String dlq, Map<String, Object> arguments) {
 
         String queueName = queueNamePrefix + "_" + routingKey;
         removeQueueFromListener(listenerId, queueName);
-        Binding binding = getBinding(exchangeName, routingKey, dlq, queueName, arguments);
+        Binding binding = getBinding(exchangeName, routingKey, queueName);
         rabbitAdmin.removeBinding(binding);
-        rabbitAdmin.deleteQueue(queueName);
+        log.debug("Removed binding");
-
+        if (rabbitAdmin.deleteQueue(queueName)) {
-    }
+            log.info("Deleted queue {} successfully", queueName);
-
+        } else {
-
+            log.info("Queue deletion failed");
-    private static Binding getBinding(String exchangeName, String routingKey, String dlq, String queueName, Map<String, Object> arguments) {
-
-        Binding binding = new Binding(queueName, Binding.DestinationType.QUEUE, exchangeName, routingKey, null);
-        if (dlq != null && !dlq.isBlank()) {
-            binding.addArgument("x-dead-letter-exchange", "");
-            binding.addArgument("x-dead-letter-routing-key", dlq);
-            arguments.forEach(binding::addArgument);
         }
-        return binding;
+
     }
 
 
     @Override
     public void addQueueToListener(String listenerId, String queueName) {
 
-        log.info("Adding queue " + queueName + " to listener " + listenerId);
+        log.info("Adding queue {} to listener {}", queueName, listenerId);
         if (!checkQueueExistOnListener(listenerId, queueName)) {
             getMessageListenerContainerById(listenerId).addQueueNames(queueName);
-        } 
+        }
     }
 
 
     @Override
     public void removeQueueFromListener(String listenerId, String queueName) {
 
-        log.info("Removing queue " + queueName + " from listener " + listenerId);
+        log.info("Removing queue {} from listener {}", queueName, listenerId);
         if (checkQueueExistOnListener(listenerId, queueName)) {
-            
+
             getMessageListenerContainerById(listenerId).removeQueueNames(queueName);
             rabbitAdmin.deleteQueue(queueName);
         }
@@ -85,23 +107,16 @@ public class RabbitQueueFromExchangeServiceImpl implements RabbitQueueFromExchan
     @Override
     public boolean checkQueueExistOnListener(String listenerId, String queueName) {
 
-        log.debug("Checking if queue " + queueName + " exists on listener " + listenerId);
+        log.debug("Checking if queue {} exists on listener {}", queueName, listenerId);
         String[] queueNames = getMessageListenerContainerById(listenerId).getQueueNames();
         for (String name : queueNames) {
             if (name.equals(queueName)) {
-                log.debug("Queue with name : " + queueName + " already exists on listener");
+                log.debug("Queue with name : {} already exists on listener", queueName);
                 return true;
             }
         }
-        log.debug("Queue with name : " + queueName + " does not exist on listener");
+        log.debug("Queue with name : {} does not exist on listener", queueName);
         return false;
     }
 
-
-    private AbstractMessageListenerContainer getMessageListenerContainerById(String listenerId) {
-
-        log.debug("Getting message listener container by id " + listenerId);
-        return ((AbstractMessageListenerContainer) rabbitListenerEndpointRegistry.getListenerContainer(listenerId));
-    }
-
 }

src/main/java/com/knecon/fforesight/tenantcommons/queue/TenantEventsLock.java

@@ -0,0 +1,23 @@
+package com.knecon.fforesight.tenantcommons.queue;
+
+import java.util.concurrent.Semaphore;
+
+import lombok.SneakyThrows;
+
+public class TenantEventsLock {
+
+    private final static Semaphore semaphore = new Semaphore(1);
+
+
+    @SneakyThrows
+    public static void executeInLock(Runnable runnable) {
+
+        try {
+            semaphore.acquire();
+            runnable.run();
+        } finally {
+            semaphore.release();
+        }
+    }
+
+}

src/main/java/com/knecon/fforesight/tenantcommons/queue/TenantExchangeMessageReceiver.java

@@ -1,70 +0,0 @@
-package com.knecon.fforesight.tenantcommons.queue;
-
-import java.util.Set;
-
-import com.knecon.fforesight.tenantcommons.TenantProvider;
-import com.knecon.fforesight.tenantcommons.model.TenantCreatedEvent;
-import com.knecon.fforesight.tenantcommons.model.TenantQueueConfiguration;
-import com.knecon.fforesight.tenantcommons.model.TenantResponse;
-
-import lombok.AccessLevel;
-import lombok.RequiredArgsConstructor;
-import lombok.experimental.FieldDefaults;
-import lombok.extern.slf4j.Slf4j;
-
-@Slf4j
-@RequiredArgsConstructor
-@FieldDefaults(makeFinal = true, level = AccessLevel.PRIVATE)
-public abstract class TenantExchangeMessageReceiver {
-
-    RabbitQueueFromExchangeService rabbitQueueService;
-    TenantProvider tenantProvider;
-
-
-    protected abstract Set<TenantQueueConfiguration> getTenantQueueConfigs();
-
-
-    public void initializeQueues() {
-
-        log.info("Initializing queues for all tenants.");
-        tenantProvider.getTenants()
-                .forEach(tenant -> {
-                    getTenantQueueConfigs().parallelStream()
-                            .forEach(tqc -> rabbitQueueService.addNewQueue(tqc.getListenerId(),
-                                                                           tqc.getQueuePrefix(),
-                                                                           tqc.getExchangeName(),
-                                                                           tenant.getTenantId(),
-                                                                           tqc.getDlqName(),
-                                                                           tqc.getArguments()));
-                    log.info("Initialized queues for tenant " + tenant.getTenantId());
-                });
-    }
-
-
-    public void reactToTenantCreation(TenantCreatedEvent tenantCreatedEvent) {
-
-        String tenantId = tenantCreatedEvent.getTenantId();
-        log.info("Creating queues for new tenant " + tenantId);
-        getTenantQueueConfigs().forEach(tqc -> rabbitQueueService.addNewQueue(tqc.getListenerId(),
-                                                                              tqc.getQueuePrefix(),
-                                                                              tqc.getExchangeName(),
-                                                                              tenantId,
-                                                                              tqc.getDlqName(),
-                                                                              tqc.getArguments()));
-    }
-
-
-    public void reactToTenantDeletion(TenantResponse tenantResponse) {
-
-        String tenantId = tenantResponse.getTenantId();
-        log.info("Removing queues for deleted tenant " + tenantId);
-        getTenantQueueConfigs().forEach(tqc -> rabbitQueueService.deleteQueue(tqc.getListenerId(),
-                                                                              tqc.getQueuePrefix(),
-                                                                              tqc.getExchangeName(),
-                                                                              tenantId,
-                                                                              tqc.getDlqName(),
-                                                                              tqc.getArguments()));
-
-    }
-
-}

src/main/java/com/knecon/fforesight/tenantcommons/queue/TenantMessagingConfiguration.java

@@ -1,29 +1,53 @@
 package com.knecon.fforesight.tenantcommons.queue;
 
+import static com.knecon.fforesight.tenantcommons.queue.TenantRabbitListener.TENANT_CREATED_LISTENER;
+import static com.knecon.fforesight.tenantcommons.queue.TenantRabbitListener.TENANT_DELETED_LISTENER;
+import static com.knecon.fforesight.tenantcommons.queue.TenantRabbitListener.TENANT_SYNC_LISTENER;
+
 import org.springframework.amqp.core.Binding;
 import org.springframework.amqp.core.BindingBuilder;
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.core.QueueBuilder;
 import org.springframework.amqp.core.TopicExchange;
+import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistry;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.context.event.ApplicationReadyEvent;
 import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.event.EventListener;
 
-import lombok.NoArgsConstructor;
+import lombok.RequiredArgsConstructor;
 
-@NoArgsConstructor
+@Configuration
-public abstract class TenantMessagingConfiguration {
+@RequiredArgsConstructor
+public class TenantMessagingConfiguration {
 
-    @Value("${POD_NAME:}")
+    public static final String TENANT_CREATED_QUEUE = "tenantCreatedQueue";
-    private String podName;
+    public static final String TENANT_DELETED_QUEUE = "tenantDeletedQueue";
-
+    public static final String TENANT_SYNC_QUEUE = "tenantSyncQueue";
-    private static final String TENANT_EVENTS_DLQ_SUFFIX = "_tenant_events_dlq";
-    private static final String TENANT_CREATED_QUEUE_SUFFIX = "_tenant_created_queue";
-    private static final String TENANT_DELETED_QUEUE_SUFFIX = "_tenant_deleted_queue";
 
+    public static final String TENANT_EXCHANGE = "tenantExchange";
     // time in ms after which a deletion will be executed when no consumer is present
     // see: https://www.rabbitmq.com/docs/ttl#queue-ttl
     public static final int QUEUE_EXPIRATION_TIME = 300000; // 5 minutes
+    private static final String TENANT_EVENTS_DLQ_SUFFIX = "_tenant_events_error";
+    private static final String TENANT_CREATED_QUEUE_SUFFIX = "_tenant_created";
+    private static final String TENANT_DELETED_QUEUE_SUFFIX = "_tenant_deleted";
+    private static final String TENANT_SYNC_QUEUE_SUFFIX = "_tenant_sync";
+    private final RabbitListenerEndpointRegistry rabbitListenerEndpointRegistry;
+    @Value("${POD_NAME:}")
+    private String podName;
+
+
+    @EventListener(ApplicationReadyEvent.class)
+    public void assignQueuesToListeners() {
+
+        rabbitListenerEndpointRegistry.getListenerContainer(TENANT_CREATED_LISTENER).setQueueNames(getTenantCreatedQueueName());
+        rabbitListenerEndpointRegistry.getListenerContainer(TENANT_DELETED_LISTENER).setQueueNames(getTenantDeletedQueueName());
+        rabbitListenerEndpointRegistry.getListenerContainer(TENANT_SYNC_LISTENER).setQueueNames(getTenantSyncQueueName());
+
+    }
 
 
     public String getTenantCreatedQueueName() {
@@ -38,6 +62,48 @@ public abstract class TenantMessagingConfiguration {
     }
 
 
+    public String getTenantSyncQueueName() {
+
+        return this.getQueueNameWithSuffix(TENANT_SYNC_QUEUE_SUFFIX);
+    }
+
+
+    @Bean(name = TENANT_EXCHANGE)
+    public TopicExchange tenantExchange(@Value("${fforesight.tenant-exchange.name:tenants-exchange}") String tenantExchangeName) {
+
+        return new TopicExchange(tenantExchangeName);
+    }
+
+
+    @Bean(TENANT_CREATED_QUEUE)
+    public Queue tenantCreatedQueue() {
+
+        return QueueBuilder.durable(getTenantCreatedQueueName())
+                .withArgument("x-dead-letter-exchange", "")
+                .withArgument("x-dead-letter-routing-key", getTenantEventsDLQName())
+                .withArgument("x-expires", QUEUE_EXPIRATION_TIME)
+                .build();
+    }
+
+
+    @Bean(TENANT_DELETED_QUEUE)
+    public Queue tenantDeletedQueue() {
+
+        return QueueBuilder.durable(getTenantDeletedQueueName())
+                .withArgument("x-dead-letter-exchange", "")
+                .withArgument("x-dead-letter-routing-key", getTenantEventsDLQName())
+                .withArgument("x-expires", QUEUE_EXPIRATION_TIME)
+                .build();
+    }
+
+
+    @Bean
+    public Queue tenantEventsDLQ() {
+
+        return QueueBuilder.durable(getTenantEventsDLQName()).withArgument("x-expires", QUEUE_EXPIRATION_TIME).build();
+    }
+
+
     public String getTenantEventsDLQName() {
 
         return getQueueNameWithSuffix(TENANT_EVENTS_DLQ_SUFFIX);
@@ -68,53 +134,36 @@ public abstract class TenantMessagingConfiguration {
     }
 
 
-    @Bean(name = "tenantExchange")
-    public TopicExchange tenantExchange(@Value("${fforesight.tenant-exchange.name:tenants-exchange}") String tenantExchangeName) {
-
-        return new TopicExchange(tenantExchangeName);
-    }
-
-
-    @Bean("tenantCreatedQueue")
-    public Queue tenantCreatedQueue() {
-
-        return QueueBuilder.durable(getTenantCreatedQueueName())
-                .withArgument("x-dead-letter-exchange", "")
-                .withArgument("x-dead-letter-routing-key", getTenantEventsDLQName())
-                .withArgument("x-expires", QUEUE_EXPIRATION_TIME)
-                .build();
-    }
-
-
-    @Bean("tenantDeletedQueue")
-    public Queue tenantDeletedQueue() {
-
-        return QueueBuilder.durable(getTenantDeletedQueueName())
-                .withArgument("x-dead-letter-exchange", "")
-                .withArgument("x-dead-letter-routing-key", getTenantEventsDLQName())
-                .withArgument("x-expires", QUEUE_EXPIRATION_TIME)
-                .build();
-    }
-
-
     @Bean
-    public Queue tenantEventsDLQ() {
+    public Binding tenantCreatedBinding(@Qualifier(TENANT_CREATED_QUEUE) Queue tenantCreatedQueue, @Qualifier(TENANT_EXCHANGE) TopicExchange tenantExchange) {
-
-        return QueueBuilder.durable(getTenantEventsDLQName()).withArgument("x-expires", QUEUE_EXPIRATION_TIME).build();
-    }
-
-
-    @Bean
-    public Binding tenantCreatedBinding(@Qualifier("tenantCreatedQueue") Queue tenantCreatedQueue, @Qualifier("tenantExchange") TopicExchange tenantExchange) {
 
         return BindingBuilder.bind(tenantCreatedQueue).to(tenantExchange).with("tenant.created");
+
     }
 
 
     @Bean
-    public Binding tenantDeletedBinding(@Qualifier("tenantDeletedQueue") Queue tenantDeletedQueue, @Qualifier("tenantExchange") TopicExchange tenantExchange) {
+    public Binding tenantDeletedBinding(@Qualifier(TENANT_DELETED_QUEUE) Queue tenantDeletedQueue, @Qualifier(TENANT_EXCHANGE) TopicExchange tenantExchange) {
 
         return BindingBuilder.bind(tenantDeletedQueue).to(tenantExchange).with("tenant.delete");
     }
 
+
+    @Bean({TENANT_SYNC_QUEUE})
+    public Queue tenantSyncQueue() {
+
+        return QueueBuilder.durable(this.getTenantSyncQueueName())
+                .withArgument("x-dead-letter-exchange", "")
+                .withArgument("x-dead-letter-routing-key", this.getTenantEventsDLQName())
+                .withArgument("x-expires", 300000)
+                .build();
+    }
+
+
+    @Bean
+    public Binding tenantSyncBinding(@Qualifier(TENANT_SYNC_QUEUE) Queue tenantCreatedQueue, @Qualifier(TENANT_EXCHANGE) TopicExchange tenantExchange) {
+
+        return BindingBuilder.bind(tenantCreatedQueue).to(tenantExchange).with("tenant.sync");
+    }
+
 }

src/main/java/com/knecon/fforesight/tenantcommons/queue/TenantQueueInitializer.java

@@ -0,0 +1,54 @@
+package com.knecon.fforesight.tenantcommons.queue;
+
+import java.util.Optional;
+
+import org.springframework.boot.context.event.ApplicationReadyEvent;
+import org.springframework.context.event.EventListener;
+import org.springframework.stereotype.Service;
+
+import com.knecon.fforesight.tenantcommons.TenantProvider;
+import com.knecon.fforesight.tenantcommons.model.TenantQueueProvider;
+import lombok.RequiredArgsConstructor;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@RequiredArgsConstructor
+@Service
+public class TenantQueueInitializer {
+
+    private final Optional<TenantQueueProvider> tenantQueueProvider;
+    private final TenantProvider tenantProvider;
+    private final RabbitQueueFromExchangeService rabbitQueueService;
+
+
+    @EventListener(ApplicationReadyEvent.class)
+    public void onApplicationReady() {
+
+        initializeQueues();
+    }
+
+
+    @SneakyThrows
+    public void initializeQueues() {
+
+        tenantQueueProvider.ifPresent(tenantQueueProvider -> {
+            log.info("Initializing queues for all tenants.");
+            TenantEventsLock.executeInLock(() -> {
+                tenantProvider.getTenants()
+                        .forEach(tenant -> {
+                            tenantQueueProvider.getTenantQueueConfigurations().parallelStream()
+                                    .forEach(tqc -> rabbitQueueService.addNewQueue(tqc.getListenerId(),
+                                            tqc.getQueuePrefix(),
+                                            tqc.getExchangeName(),
+                                            tenant.getTenantId(),
+                                            tqc.getDlqName(),
+                                            tqc.getArguments()));
+                            log.info("Initialized queues for tenant {}", tenant.getTenantId());
+                        });
+            });
+        });
+
+    }
+
+}

src/main/java/com/knecon/fforesight/tenantcommons/queue/TenantRabbitListener.java

@@ -0,0 +1,83 @@
+package com.knecon.fforesight.tenantcommons.queue;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.springframework.amqp.rabbit.annotation.RabbitHandler;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.stereotype.Service;
+
+import com.knecon.fforesight.tenantcommons.listener.ITenantEventHandler;
+import com.knecon.fforesight.tenantcommons.model.ITenantEvent;
+import com.knecon.fforesight.tenantcommons.model.TenantCreatedEvent;
+import com.knecon.fforesight.tenantcommons.model.TenantResponse;
+import com.knecon.fforesight.tenantcommons.model.TenantSyncEvent;
+import lombok.RequiredArgsConstructor;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class TenantRabbitListener {
+
+    public static final String TENANT_CREATED_LISTENER = "tenantCreatedListener";
+    public static final String TENANT_DELETED_LISTENER = "tenantDeletedListener";
+    public static final String TENANT_SYNC_LISTENER = "tenantSyncListener";
+    private final List<ITenantEventHandler> eventListeners;
+
+
+    @SneakyThrows
+    @RabbitHandler
+    @RabbitListener(id = TENANT_CREATED_LISTENER)
+    public void listenOnTenantCreated(TenantCreatedEvent tenantCreatedEvent) {
+
+        this.processTenantEvent(tenantCreatedEvent);
+
+    }
+
+
+    @SneakyThrows
+    private <T extends ITenantEvent> void processTenantEvent(T event) {
+
+        TenantEventsLock.executeInLock(() -> {
+            List<Throwable> exceptions = new ArrayList<>();
+            eventListeners.stream()
+                    .filter(listener -> listener.getEventClass().equals(event.getClass()))
+                    .forEach(listener -> {
+                        log.info("Processing listener {} for {} event {}", listener.getClass().getSimpleName(), event.getClass().getSimpleName(), event.getTenantId());
+                        try {
+                            listener.handle(event);
+                        } catch (Exception e) {
+                            log.error("Error handling event {} at listener: {}", event.getTenantId(), listener.getEventClass().getSimpleName(), e);
+                            exceptions.add(e);
+                        }
+                    });
+            if (!exceptions.isEmpty()) {
+                log.error("Some listeners ({}) failed to handle tenant created event {}", exceptions.size(), event.getTenantId());
+                // TODO determine if we should do something if some listeners failed ...
+            }
+        });
+    }
+
+
+    @SneakyThrows
+    @RabbitHandler
+    @RabbitListener(id = TENANT_DELETED_LISTENER)
+    public void listenOnTenantDeleted(TenantResponse tenantResponse) {
+
+        this.processTenantEvent(tenantResponse);
+
+    }
+
+
+    @SneakyThrows
+    @RabbitHandler
+    @RabbitListener(id = TENANT_SYNC_LISTENER)
+    public void listenOnTenantSync(TenantSyncEvent tenantSyncEvent) {
+
+        this.processTenantEvent(tenantSyncEvent);
+
+    }
+
+}

src/main/java/com/knecon/fforesight/tenantcommons/task/ApplicationTaskExecutorBeanConfig.java

@@ -4,6 +4,7 @@ import static org.springframework.boot.autoconfigure.task.TaskExecutionAutoConfi
 import static org.springframework.scheduling.annotation.AsyncAnnotationBeanPostProcessor.DEFAULT_TASK_EXECUTOR_BEAN_NAME;
 
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.task.TaskExecutorBuilder;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -16,8 +17,10 @@ public class ApplicationTaskExecutorBeanConfig {
 
     @Lazy
     @Bean(name = {APPLICATION_TASK_EXECUTOR_BEAN_NAME, DEFAULT_TASK_EXECUTOR_BEAN_NAME})
+    @ConditionalOnMissingBean(name = {APPLICATION_TASK_EXECUTOR_BEAN_NAME, DEFAULT_TASK_EXECUTOR_BEAN_NAME})
     public ThreadPoolTaskExecutor applicationTaskExecutor(TaskExecutorBuilder builder) {
+
         return builder.build();
     }
 
-}
+}

src/main/java/com/knecon/fforesight/tenantcommons/task/KneconTaskDecorator.java

@@ -4,4 +4,5 @@ import org.springframework.core.task.TaskDecorator;
 
 // Marker interface
 public interface KneconTaskDecorator extends TaskDecorator {
+
 }

src/main/java/com/knecon/fforesight/tenantcommons/task/TaskDecoratorConfiguration.java

@@ -1,10 +1,11 @@
 package com.knecon.fforesight.tenantcommons.task;
 
-import com.knecon.fforesight.tenantcommons.TenantAwareTaskDecorator;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
+import com.knecon.fforesight.tenantcommons.TenantAwareTaskDecorator;
+
 @Configuration(proxyBeanMethods = false)
 public class TaskDecoratorConfiguration {
 
@@ -14,4 +15,5 @@ public class TaskDecoratorConfiguration {
 
         return new TenantAwareTaskDecorator();
     }
+
 }

src/main/java/com/knecon/fforesight/tenantcommons/task/TaskExecutionConfiguration.java

@@ -26,8 +26,10 @@ public class TaskExecutionConfiguration {
     @Value("${spring.task.execution.pool.warnPercentageThreshold:80}")
     private int executorWarnPercentageThreshold;
 
+
     @Bean
     public TaskExecutorCustomizer taskExecutorCustomizer(ObjectProvider<KneconTaskDecorator> taskDecorator) {
+
         return taskExecutor -> {
             taskExecutor.setTaskDecorator((KneconTaskDecorator) runnable -> {
                 val taskDecoratorUsed = taskDecorator.getIfUnique(IDENTITY_KNECON_TASK_DECORATOR_SUPPLIER);
@@ -37,11 +39,12 @@ public class TaskExecutionConfiguration {
 
                 if (queueOccupancyPercentage >= executorWarnPercentageThreshold) {
                     log.warn("Executor pool [ " + taskExecutor + " ] queue size reached " + actualQueueSize + "/" + taskExecutor.getMaxPoolSize() +
-                             " entries awaiting execution triggering the warn level set for " + executorWarnPercentageThreshold +"% occupancy.");
+                             " entries awaiting execution triggering the warn level set for " + executorWarnPercentageThreshold + "% occupancy.");
                 }
 
                 return decoratedRunnable;
             });
         };
     }
+
 }

src/main/java/com/knecon/fforesight/tenantcommons/utils/MongoConnectionStringHelper.java

@@ -1,7 +1,6 @@
 package com.knecon.fforesight.tenantcommons.utils;
 
 import com.knecon.fforesight.tenantcommons.model.MongoDBConnection;
-
 import lombok.experimental.UtilityClass;
 
 @UtilityClass

src/test/java/com/knecon/fforesight/tenantcommons/BasicIntegrationTest.java

@@ -0,0 +1,81 @@
+package com.knecon.fforesight.tenantcommons;
+
+import java.util.Collections;
+
+import org.springframework.amqp.core.DirectExchange;
+import org.springframework.boot.autoconfigure.ImportAutoConfiguration;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.util.TestPropertyValues;
+import org.springframework.context.ApplicationContextInitializer;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.ContextConfiguration;
+
+import com.knecon.fforesight.tenantcommons.model.TenantQueueConfiguration;
+import com.knecon.fforesight.tenantcommons.model.TenantQueueProvider;
+import com.knecon.fforesight.tenantcommons.utils.TestMessageService;
+import lombok.extern.slf4j.Slf4j;
+import org.testcontainers.containers.RabbitMQContainer;
+
+@Slf4j
+@ContextConfiguration(initializers = {BasicIntegrationTest.Initializer.class})
+@DirtiesContext(classMode = DirtiesContext.ClassMode.BEFORE_EACH_TEST_METHOD)
+@ImportAutoConfiguration({MultiTenancyAutoConfiguration.class})
+@SuppressWarnings("PMD.TestClassWithoutTestCases")
+@SpringBootTest
+public abstract class BasicIntegrationTest {
+
+    static class Initializer implements ApplicationContextInitializer<ConfigurableApplicationContext> {
+
+        @SuppressWarnings("PMD")
+        public void initialize(ConfigurableApplicationContext configurableApplicationContext) {
+
+            RabbitMQContainer rabbitContainer = new RabbitMQContainer("rabbitmq:3.12");
+
+            rabbitContainer.start();
+            log.info("Rabbit container started and available at {}", rabbitContainer.getHttpUrl());
+
+            TestPropertyValues.of("RABBITMQ_USERNAME=" + rabbitContainer.getAdminUsername(),
+                    "RABBITMQ_PASSWORD=" + rabbitContainer.getAdminPassword(),
+                    "RABBITMQ_HOST=" + rabbitContainer.getHost(),
+                    "RABBITMQ_PORT=" + rabbitContainer.getAmqpPort(),
+                    "POD_NAME=tenant-commons"
+            ).applyTo(configurableApplicationContext.getEnvironment());
+        }
+
+    }
+
+    @SpringBootApplication
+    static class TestConfiguration {
+
+        @Bean
+        public DirectExchange testExchange() {
+
+            return new DirectExchange("testExchange");
+        }
+
+
+        @Bean
+        public TenantQueueProvider tenantQueueProvider() {
+
+            var testQueue = TenantQueueConfiguration.builder()
+                    .listenerId("testListener")
+                    .exchangeName("testExchange")
+                    .queuePrefix("test")
+                    .dlqName("test_error")
+                    .build();
+            return new TenantQueueProvider(Collections.singleton(testQueue));
+        }
+
+
+        @Bean
+        public TestMessageService testMessageService() {
+
+            return new TestMessageService();
+        }
+
+    }
+
+}

src/test/java/com/knecon/fforesight/tenantcommons/TenantQueuesIntegrationTest.java

@@ -0,0 +1,79 @@
+package com.knecon.fforesight.tenantcommons;
+
+import static com.knecon.fforesight.tenantcommons.queue.TenantMessagingConfiguration.TENANT_EXCHANGE;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
+import static org.junit.jupiter.api.Assertions.fail;
+
+import org.springframework.amqp.rabbit.core.RabbitAdmin;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+
+import com.knecon.fforesight.tenantcommons.model.TenantCreatedEvent;
+import com.knecon.fforesight.tenantcommons.utils.TestMessage;
+import com.knecon.fforesight.tenantcommons.utils.TestMessageService;
+import com.knecon.fforesight.tenantcommons.utils.TestTenantProvider;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+import org.junit.jupiter.api.Test;
+
+public class TenantQueuesIntegrationTest extends BasicIntegrationTest {
+
+    @Autowired
+    private RabbitTemplate rabbitTemplate;
+
+    @Autowired
+    private TestTenantProvider tenantProvider;
+
+    @Autowired
+    private TestMessageService testMessageService;
+
+    @Autowired
+    private RabbitAdmin rabbitAdmin;
+
+    @Value("${fforesight.tenant-exchange.name:tenants-exchange}")
+    String tenantExchangeName;
+
+
+    @Test
+    @SneakyThrows
+    public void testTenantCreatedQueue() {
+
+        TenantContext.setTenantId("tenant1");
+        // create a new tenant
+        rabbitTemplate.convertAndSend(tenantExchangeName, "tenant.created", new TenantCreatedEvent("tenant1"));
+        // wait for message to send and be processed
+        int iterations = 0;
+        do {
+            iterations++;
+            Thread.sleep(1000);
+            if (iterations > 10) {
+                fail("Tenant not created within 10 seconds");
+            }
+            if (!tenantProvider.getTenants().isEmpty()) {
+                assertThat(tenantProvider.getTenant("tenant1")).isNotNull();
+                break;
+            }
+        } while (true);
+
+
+        rabbitTemplate.convertAndSend("testExchange", "tenant1", new TestMessage("Hello Tenant Commons!"));
+
+        iterations = 0;
+        do {
+            iterations++;
+            Thread.sleep(1000);
+            if (iterations > 10) {
+                fail("Message not sent within 10 seconds");
+            }
+            if (!testMessageService.getMessages().isEmpty()) {
+
+                break;
+            }
+        } while (true);
+
+        assertThat(testMessageService.getMessages().size()).isEqualTo(1);
+        assertThat(testMessageService.getMessages().get(0).getMessage()).isEqualTo("Hello Tenant Commons!");
+    }
+
+}

src/test/java/com/knecon/fforesight/tenantcommons/utils/TestMessage.java

@@ -0,0 +1,13 @@
+package com.knecon.fforesight.tenantcommons.utils;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+public class TestMessage {
+
+    private String message;
+}

src/test/java/com/knecon/fforesight/tenantcommons/utils/TestMessageService.java

@@ -0,0 +1,44 @@
+package com.knecon.fforesight.tenantcommons.utils;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.springframework.amqp.rabbit.annotation.RabbitHandler;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+
+import jakarta.annotation.PostConstruct;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class TestMessageService {
+
+    public List<TestMessage> messages = new ArrayList<>();
+
+
+    @PostConstruct
+    protected void init() {
+
+        log.info("TestMessageService initialized");
+    }
+
+    @RabbitHandler
+    @RabbitListener(id = "testListener")
+    public void handleTestMessage(TestMessage message) {
+
+        log.info("Received message: {}", message);
+        addMessage(message);
+    }
+
+
+    public void addMessage(TestMessage message) {
+
+        messages.add(message);
+    }
+
+
+    public List<TestMessage> getMessages() {
+
+        return messages;
+    }
+
+}

src/test/java/com/knecon/fforesight/tenantcommons/utils/TestTenantProvider.java

@@ -0,0 +1,56 @@
+package com.knecon.fforesight.tenantcommons.utils;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import org.springframework.stereotype.Service;
+
+import com.knecon.fforesight.tenantcommons.TenantProvider;
+import com.knecon.fforesight.tenantcommons.listener.ITenantEventHandler;
+import com.knecon.fforesight.tenantcommons.model.TenantCreatedEvent;
+import com.knecon.fforesight.tenantcommons.model.TenantResponse;
+import com.knecon.fforesight.tenantcommons.model.UpdateDetailsRequest;
+
+@Service
+public class TestTenantProvider implements TenantProvider, ITenantEventHandler<TenantCreatedEvent> {
+
+    private List<TenantResponse> tenants = new ArrayList<>();
+
+
+    @Override
+    public void updateDetails(String s, UpdateDetailsRequest updateDetailsRequest) {
+        //noop
+    }
+
+
+    @Override
+    public TenantResponse getTenant(String s) {
+
+        return tenants.stream().filter(t -> t.getTenantId().equals(s)).findFirst().orElse(null);
+    }
+
+
+    @Override
+    public List<TenantResponse> getTenants() {
+
+        return tenants;
+    }
+
+
+    @Override
+    public void handle(TenantCreatedEvent event) {
+
+        tenants.add(TenantResponse.builder()
+                .tenantId(event.getTenantId())
+                .build());
+    }
+
+
+    @Override
+    public Class<TenantCreatedEvent> getEventClass() {
+
+        return TenantCreatedEvent.class;
+    }
+
+}

src/test/resources/application.yaml

@@ -0,0 +1,19 @@
+spring:
+  application:
+    name: tenant-commons-tester
+  rabbitmq:
+    host: ${RABBITMQ_HOST:localhost}
+    port: ${RABBITMQ_PORT:5672}
+    username: ${RABBITMQ_USERNAME:user}
+    password: ${RABBITMQ_PASSWORD:rabbitmq}
+    listener:
+      simple:
+        acknowledge-mode: AUTO
+        concurrency: 5
+        retry:
+          enabled: true
+          max-attempts: 3
+          max-interval: 15000
+        prefetch: 1
+
+logging.type: ${LOGGING_TYPE:CONSOLE}

src/test/resources/logback-spring.xml

@@ -0,0 +1,13 @@
+<configuration>
+
+    <springProperty scope="configuration" name="logType" source="logging.type"/>
+    <springProperty scope="context" name="application.name" source="spring.application.name"/>
+    <springProperty scope="context" name="version" source="project.version"/>
+    <include resource="org/springframework/boot/logging/logback/defaults.xml"/>
+    <include resource="org/springframework/boot/logging/logback/console-appender.xml"/>
+
+    <root level="INFO">
+        <appender-ref ref="${logType}"/>
+    </root>
+
+</configuration>

src/test/resources/testcontainers.properties

@@ -0,0 +1 @@
+hub.image.name.prefix=docker-dev.knecon.com/tests/