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
@@ -10,10 +9,11 @@ plugins {
 }
 
 val storageCommonsVersion = "2.43.0"
-val springBootVersion = "3.1.4"
+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}")
-    runtimeOnly("org.springframework.boot:spring-boot-devtools:${springBootVersion}")
+    testImplementation("org.springframework.boot:spring-boot-configuration-processor:${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/MultiTenancyAutoConfiguration.java

@@ -10,7 +10,7 @@ import lombok.extern.slf4j.Slf4j;
 @Slf4j
 @AutoConfiguration
 @EnableFeignClients(basePackageClasses = TenantsClient.class)
-@ComponentScan(basePackageClasses = MultiTenancyAutoConfiguration.class)
+@ComponentScan(basePackageClasses = {MultiTenancyAutoConfiguration.class})
 public class MultiTenancyAutoConfiguration {
 
     @PostConstruct

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

@@ -1,10 +1,12 @@
 package com.knecon.fforesight.tenantcommons;
 
-import com.fasterxml.jackson.databind.DeserializationFeature;
-import com.fasterxml.jackson.databind.ObjectMapper;
+import static com.knecon.fforesight.tenantcommons.ForwardTenantInterceptor.TENANT_HEADER_NAME;
+
 import org.springframework.amqp.core.MessagePostProcessor;
 import org.springframework.amqp.rabbit.config.ContainerCustomizer;
 import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
 import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
@@ -17,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)
@@ -35,6 +38,7 @@ public class MultiTenancyMessagingConfiguration {
     @Primary
     public ContainerCustomizer<SimpleMessageListenerContainer> simpleMessageListenerContainerCustomizer(
             ObjectProvider<SimpleMessageListenerContainerCustomizer> customizers) {
+
         return container -> customizers.orderedStream().forEach((customizer) -> customizer.customize(container));
     }
 
@@ -43,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);
 
@@ -87,4 +102,12 @@ public class MultiTenancyMessagingConfiguration {
             return message;
         };
     }
+
+
+    @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/MongoDBConnection.java

@@ -0,0 +1,21 @@
+package com.knecon.fforesight.tenantcommons.model;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class MongoDBConnection {
+
+    private String prefix;
+    private String username;
+    private String password;
+    private String address;
+    private String database;
+    private String options;
+
+}

src/main/java/com/knecon/fforesight/tenantcommons/model/TenantCreatedEvent.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 TenantCreatedEvent implements ITenantEvent {
+
+    private String tenantId;
+
+}

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

@@ -0,0 +1,24 @@
+package com.knecon.fforesight.tenantcommons.model;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class TenantQueueConfiguration {
+
+    private String listenerId;
+    private String exchangeName;
+    private String queuePrefix;
+    private String dlqName;
+    @Builder.Default
+    private Map<String, Object> arguments = new HashMap<>();
+
+}

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;
@@ -22,6 +22,7 @@ public class TenantResponse {
     private SearchConnection searchConnection;
     private AzureStorageConnection azureStorageConnection;
     private S3StorageConnection s3StorageConnection;
+    private MongoDBConnection mongoDBConnection;
     private AuthDetails authDetails;
 
     private Map<String, Object> details = new HashMap<>();

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

@@ -0,0 +1,21 @@
+package com.knecon.fforesight.tenantcommons.queue;
+
+import java.util.Map;
+
+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);
+
+
+    void addQueueToListener(String listenerId, String queueName);
+
+
+    void removeQueueFromListener(String listenerId, String queueName);
+
+
+    boolean checkQueueExistOnListener(String listenerId, String queueName);
+
+}

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

@@ -0,0 +1,122 @@
+package com.knecon.fforesight.tenantcommons.queue;
+
+import java.util.Map;
+
+import org.springframework.amqp.core.Binding;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.rabbit.core.RabbitAdmin;
+import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
+import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistry;
+import org.springframework.stereotype.Service;
+
+import lombok.AccessLevel;
+import lombok.RequiredArgsConstructor;
+import lombok.SneakyThrows;
+import lombok.experimental.FieldDefaults;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@Service
+@RequiredArgsConstructor
+@FieldDefaults(makeFinal = true, level = AccessLevel.PRIVATE)
+public class RabbitQueueFromExchangeServiceImpl implements RabbitQueueFromExchangeService {
+
+    RabbitAdmin rabbitAdmin;
+    RabbitListenerEndpointRegistry rabbitListenerEndpointRegistry;
+
+
+    @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);
+        addArgumentsToQueue(queue, dlq, arguments);
+        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, queueName);
+        rabbitAdmin.removeBinding(binding);
+        log.debug("Removed binding");
+        if (rabbitAdmin.deleteQueue(queueName)) {
+            log.info("Deleted queue {} successfully", queueName);
+        } else {
+            log.info("Queue deletion failed");
+        }
+
+    }
+
+
+    @Override
+    public void addQueueToListener(String listenerId, String queueName) {
+
+        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 {} from listener {}", queueName, listenerId);
+        if (checkQueueExistOnListener(listenerId, queueName)) {
+
+            getMessageListenerContainerById(listenerId).removeQueueNames(queueName);
+            rabbitAdmin.deleteQueue(queueName);
+        }
+    }
+
+
+    @Override
+    public boolean checkQueueExistOnListener(String listenerId, String queueName) {
+
+        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 : {} already exists on listener", queueName);
+                return true;
+            }
+        }
+        log.debug("Queue with name : {} does not exist on listener", queueName);
+        return false;
+    }
+
+}

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/TenantMessagingConfiguration.java

@@ -0,0 +1,169 @@
+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.RequiredArgsConstructor;
+
+@Configuration
+@RequiredArgsConstructor
+public class TenantMessagingConfiguration {
+
+    public static final String TENANT_CREATED_QUEUE = "tenantCreatedQueue";
+    public static final String TENANT_DELETED_QUEUE = "tenantDeletedQueue";
+    public static final String TENANT_SYNC_QUEUE = "tenantSyncQueue";
+
+    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() {
+
+        return getQueueNameWithSuffix(TENANT_CREATED_QUEUE_SUFFIX);
+    }
+
+
+    public String getTenantDeletedQueueName() {
+
+        return getQueueNameWithSuffix(TENANT_DELETED_QUEUE_SUFFIX);
+    }
+
+
+    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);
+    }
+
+
+    protected String getQueueNameWithSuffix(String suffix) {
+
+        if (!useDefaultQueueName() && podName != null && !podName.isEmpty()) {
+            return podName + suffix;
+        } else {
+            return getDefaultQueueName(suffix);
+        }
+    }
+
+
+    protected boolean useDefaultQueueName() {
+
+        return false;
+    }
+
+
+    // This method will be overridden by subclasses if the POD_NAME is not set.
+    // Default implementation throws an exception to ensure it's implemented if used.
+    protected String getDefaultQueueName(String suffix) {
+
+        throw new UnsupportedOperationException("Queue name method not implemented");
+    }
+
+
+    @Bean
+    public Binding tenantCreatedBinding(@Qualifier(TENANT_CREATED_QUEUE) Queue tenantCreatedQueue, @Qualifier(TENANT_EXCHANGE) TopicExchange tenantExchange) {
+
+        return BindingBuilder.bind(tenantCreatedQueue).to(tenantExchange).with("tenant.created");
+
+    }
+
+
+    @Bean
+    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

@@ -0,0 +1,26 @@
+package com.knecon.fforesight.tenantcommons.task;
+
+import static org.springframework.boot.autoconfigure.task.TaskExecutionAutoConfiguration.APPLICATION_TASK_EXECUTOR_BEAN_NAME;
+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;
+import org.springframework.context.annotation.Lazy;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+
+@ConditionalOnClass(ThreadPoolTaskExecutor.class)
+@Configuration
+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

@@ -1,11 +1,50 @@
 package com.knecon.fforesight.tenantcommons.task;
 
+import java.util.function.Supplier;
+
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.task.TaskExecutorCustomizer;
+import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.PropertySource;
+import org.springframework.scheduling.annotation.EnableAsync;
+
+import lombok.extern.slf4j.Slf4j;
+import lombok.val;
 
 @Configuration(proxyBeanMethods = false)
 @PropertySource("classpath:tenant-commons-task.properties")
 @ConditionalOnProperty(prefix = "management.tracing", name = "enabled", havingValue = "false")
+@EnableAsync
+@Slf4j
 public class TaskExecutionConfiguration {
+
+    private final static Supplier<KneconTaskDecorator> IDENTITY_KNECON_TASK_DECORATOR_SUPPLIER = () -> runna -> runna;
+
+    @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);
+                val decoratedRunnable = taskDecoratorUsed.decorate(runnable);
+                val actualQueueSize = taskExecutor.getQueueSize() + 1;
+                val queueOccupancyPercentage = actualQueueSize * 100.f / taskExecutor.getMaxPoolSize();
+
+                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.");
+                }
+
+                return decoratedRunnable;
+            });
+        };
+    }
+
 }

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

@@ -0,0 +1,44 @@
+package com.knecon.fforesight.tenantcommons.utils;
+
+import com.knecon.fforesight.tenantcommons.model.MongoDBConnection;
+import lombok.experimental.UtilityClass;
+
+@UtilityClass
+public class MongoConnectionStringHelper {
+
+    public static String buildGenericMongoUrl(MongoDBConnection mongoDBConnection) {
+
+        return mongoDBConnection.getPrefix() + "://" + mongoDBConnection.getAddress() + "/" + mongoDBConnection.getDatabase() + "?" + mongoDBConnection.getOptions();
+    }
+
+
+    public static String buildGenericMongoConnectionString(MongoDBConnection mongoDBConnection) {
+
+        return mongoDBConnection.getPrefix()
+               + "://"
+               + mongoDBConnection.getUsername()
+               + ":"
+               + mongoDBConnection.getPassword()
+               + "@"
+               + mongoDBConnection.getAddress()
+               + "/?"
+               + mongoDBConnection.getOptions();
+    }
+
+
+    public static String buildDatabaseMongoConnectionString(MongoDBConnection mongoDBConnection) {
+
+        return mongoDBConnection.getPrefix()
+               + "://"
+               + mongoDBConnection.getUsername()
+               + ":"
+               + mongoDBConnection.getPassword()
+               + "@"
+               + mongoDBConnection.getAddress()
+               + "/"
+               + mongoDBConnection.getDatabase()
+               + "?"
+               + mongoDBConnection.getOptions();
+    }
+
+}

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/