diff --git a/dc3-common/dc3-common-mq-kafka/pom.xml b/dc3-common/dc3-common-mq-kafka/pom.xml new file mode 100644 index 000000000..cdab9f459 --- /dev/null +++ b/dc3-common/dc3-common-mq-kafka/pom.xml @@ -0,0 +1,51 @@ + + + + 4.0.0 + + + io.github.pnoker + dc3-common + 2026.5.22 + + + ${project.artifactId} + dc3-common-mq-kafka + 2026.5.22 + jar + + IoT DC3 Kafka adapter for the broker-neutral messaging port + + + + org.springframework.kafka + spring-kafka + + + io.github.pnoker + dc3-common-mq + + + io.github.pnoker + dc3-common-constant + + + + diff --git a/dc3-common/dc3-common-mq-kafka/src/main/java/io/github/pnoker/common/mq/kafka/KafkaMqAdapter.java b/dc3-common/dc3-common-mq-kafka/src/main/java/io/github/pnoker/common/mq/kafka/KafkaMqAdapter.java new file mode 100644 index 000000000..3480034a6 --- /dev/null +++ b/dc3-common/dc3-common-mq-kafka/src/main/java/io/github/pnoker/common/mq/kafka/KafkaMqAdapter.java @@ -0,0 +1,361 @@ +/* + * Copyright 2016-present the IoT DC3 original author or authors. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +package io.github.pnoker.common.mq.kafka; + +import io.github.pnoker.common.constant.mq.ConsumptionProfile; +import io.github.pnoker.common.constant.mq.MqTopic; +import io.github.pnoker.common.constant.mq.OrderingGuarantee; +import io.github.pnoker.common.constant.mq.SubscriptionMode; +import io.github.pnoker.common.mq.adapter.BrokerAdapter; +import io.github.pnoker.common.mq.adapter.BrokerCapabilities; +import io.github.pnoker.common.mq.adapter.RawBatchListener; +import io.github.pnoker.common.mq.adapter.RawDeliveryListener; +import io.github.pnoker.common.mq.adapter.WireConfirmation; +import io.github.pnoker.common.mq.adapter.WireMqDelivery; +import io.github.pnoker.common.mq.config.BatchConsumerProperties; +import io.github.pnoker.common.mq.listener.MqPoisonException; +import io.github.pnoker.common.mq.message.WireMqMessage; +import io.github.pnoker.common.mq.subscription.SubscriptionSpec; +import lombok.extern.slf4j.Slf4j; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.header.Header; +import org.apache.kafka.common.serialization.ByteArrayDeserializer; +import org.apache.kafka.common.serialization.ByteArraySerializer; +import org.apache.kafka.common.serialization.StringDeserializer; +import org.apache.kafka.common.serialization.StringSerializer; +import org.springframework.kafka.core.ConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.listener.BatchAcknowledgingMessageListener; +import org.springframework.kafka.listener.AcknowledgingMessageListener; +import org.springframework.kafka.listener.MessageListenerContainer; +import org.springframework.kafka.listener.ConcurrentMessageListenerContainer; +import org.springframework.kafka.listener.ContainerProperties; +import org.springframework.kafka.support.Acknowledgment; + +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.UUID; +import java.util.concurrent.CopyOnWriteArrayList; + +/** + * Kafka implementation of the broker port. Topics map to {@code dc3.} (logical + * dead-letter topics map to {@code dc3..dlq}, so rejects and dead-letter + * subscriptions land on the same topic); the partition key becomes the record key + * (per-key ordering via partitioning); LOAD_BALANCE rides consumer groups while + * BROADCAST uses per-instance group ids. Delayed delivery is delegated to the port's + * local-scheduler fallback (capability false); rejecting without requeue republishes + * the record(s) to the {@code .dlq} topic and commits. + * + * @author pnoker + * @since 2026.8.19 + */ +@Slf4j +public class KafkaMqAdapter implements BrokerAdapter { + + /** + * Physical topic prefix; the design doc's namespace knob lands here in phase 3. + */ + private static final String TOPIC_PREFIX = "dc3."; + + private final KafkaTemplate kafkaTemplate; + private final Map baseConsumerConfig; + private final BatchConsumerProperties retryProperties; + + private final List containers = new CopyOnWriteArrayList<>(); + + public KafkaMqAdapter(KafkaTemplate kafkaTemplate, Map baseConsumerConfig, + BatchConsumerProperties retryProperties) { + this.kafkaTemplate = kafkaTemplate; + this.baseConsumerConfig = baseConsumerConfig; + this.retryProperties = retryProperties; + } + + /** + * Build a byte-array template against the given bootstrap servers ({@code acks=all} + * so completed send futures mean the broker accepted the record). + * + * @param bootstrapServers kafka bootstrap servers + * @return configured template + */ + public static KafkaTemplate template(String bootstrapServers) { + Map producer = new HashMap<>(); + producer.put(org.apache.kafka.clients.producer.ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); + producer.put(org.apache.kafka.clients.producer.ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, + StringSerializer.class); + producer.put(org.apache.kafka.clients.producer.ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, + ByteArraySerializer.class); + producer.put(org.apache.kafka.clients.producer.ProducerConfig.ACKS_CONFIG, "all"); + producer.put(org.apache.kafka.clients.producer.ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); + DefaultKafkaProducerFactory factory = new DefaultKafkaProducerFactory<>(producer); + KafkaTemplate template = new KafkaTemplate<>(factory); + template.setDefaultTopic(TOPIC_PREFIX + "default"); + return template; + } + + /** + * Base consumer configuration (bootstrap servers only) the adapter copies per + * subscription. + * + * @param bootstrapServers kafka bootstrap servers + * @return consumer config skeleton + */ + public static Map consumerConfig(String bootstrapServers) { + Map config = new HashMap<>(); + config.put(org.apache.kafka.clients.consumer.ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); + config.put(org.apache.kafka.clients.consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, + StringDeserializer.class); + config.put(org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, + ByteArrayDeserializer.class); + config.put(org.apache.kafka.clients.consumer.ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); + // latest mirrors the rabbit fresh-queue semantics: a brand new consumer group only sees + // messages published after it joins, instead of replaying the topic backlog. + config.put(org.apache.kafka.clients.consumer.ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest"); + return config; + } + + @Override + public String type() { + return "kafka"; + } + + @Override + public BrokerCapabilities capabilities() { + return new BrokerCapabilities(false, false, true, false, true, true, false, OrderingGuarantee.PER_KEY); + } + + @Override + public void publish(WireMqMessage message) { + kafkaTemplate.send(producerRecord(message)); + } + + @Override + public void publish(WireMqMessage message, WireConfirmation confirmation) { + kafkaTemplate.send(producerRecord(message)) + .whenComplete((result, failure) -> confirmation.onConfirm(message, Objects.isNull(failure), failure)); + } + + @Override + public void subscribe(SubscriptionSpec spec, RawDeliveryListener listener) { + ConcurrentMessageListenerContainer container = + new ConcurrentMessageListenerContainer<>(consumerFactory(spec, false), + containerProperties(spec)); + container.getContainerProperties().setMessageListener( + (AcknowledgingMessageListener) (record, springAck) -> { + try { + listener.onDelivery(deliveryOf(record, List.of(record), springAck)); + } catch (MqPoisonException e) { + publishDead(record); + springAck.acknowledge(); + } catch (Exception e) { + log.warn("Kafka delivery failed, nacking for redelivery, topic={}, offset={}", + record.topic(), record.offset(), e); + springAck.nack(Duration.ofMillis(50)); + } + }); + start(spec, container); + } + + @Override + public void subscribeBatch(SubscriptionSpec spec, RawBatchListener listener) { + ConcurrentMessageListenerContainer container = + new ConcurrentMessageListenerContainer<>(consumerFactory(spec, true), + containerProperties(spec)); + container.getContainerProperties().setMessageListener( + (BatchAcknowledgingMessageListener) (records, springAck) -> { + List> batch = new ArrayList<>(); + records.forEach(batch::add); + if (batch.isEmpty()) { + return; + } + // Synchronous bounded retry with backoff, mirroring the rabbit batch + // factory's stateless retry advice; exhaustion dead-letters the whole + // batch and commits instead of dropping it silently. + int maxAttempts = Math.max(1, retryProperties.getMaxRetries()) + 1; + for (int attempt = 1; ; attempt++) { + try { + listener.onBatch(batch.stream() + .map(record -> deliveryOf(record, batch, springAck)).toList()); + return; + } catch (MqPoisonException e) { + log.warn("Kafka poison batch dead-lettered, size={}", batch.size(), e); + batch.forEach(this::publishDead); + springAck.acknowledge(); + return; + } catch (Exception e) { + if (attempt >= maxAttempts) { + log.error("Kafka batch exhausted retries, dead-lettering, size={}", + batch.size(), e); + batch.forEach(this::publishDead); + springAck.acknowledge(); + return; + } + sleepBackoff(attempt); + } + } + }); + start(spec, container); + } + + /** + * Exponential backoff between synchronous batch retry attempts, bounded by the + * configured ceiling; interrupted sleep aborts the wait without failing the batch. + */ + private void sleepBackoff(int attempt) { + long initial = retryProperties.getRetryInitialIntervalMillis(); + long cap = retryProperties.getRetryMaxIntervalMillis(); + long exponent = Math.min(Math.max(attempt - 1, 0), 30); + long delay = initial >= cap ? cap : Math.min(initial * (1L << exponent), cap); + try { + Thread.sleep(delay); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + + /** + * Stop every container this adapter started. + */ + public void stop() { + containers.forEach(MessageListenerContainer::stop); + containers.clear(); + } + + private void start(SubscriptionSpec spec, ConcurrentMessageListenerContainer container) { + container.setAutoStartup(true); + container.setConcurrency(spec.profile() == ConsumptionProfile.THROUGHPUT ? 4 : 2); + containers.add(container); + container.start(); + log.info("Kafka subscription started, topic={}, mode={}, delivery={}, groupId={}", + spec.topic(), spec.mode(), spec.delivery(), groupIdOf(spec)); + } + + private ConsumerFactory consumerFactory(SubscriptionSpec spec, boolean batch) { + Map props = new HashMap<>(baseConsumerConfig); + props.put(org.apache.kafka.clients.consumer.ConsumerConfig.GROUP_ID_CONFIG, groupIdOf(spec)); + if (batch) { + props.put(org.apache.kafka.clients.consumer.ConsumerConfig.MAX_POLL_RECORDS_CONFIG, + Math.max(1, retryProperties.getBatchSize())); + } + return new DefaultKafkaConsumerFactory<>(props); + } + + private ContainerProperties containerProperties(SubscriptionSpec spec) { + ContainerProperties properties = new ContainerProperties(topicName(spec.topic())); + properties.setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE); + properties.setGroupId(groupIdOf(spec)); + return properties; + } + + + private String groupIdOf(SubscriptionSpec spec) { + String base = spec.group().isBlank() ? TOPIC_PREFIX + spec.topic().name().toLowerCase() : spec.group(); + if (spec.mode() == SubscriptionMode.BROADCAST) { + return base + "-" + UUID.randomUUID(); + } + return base; + } + + private WireMqDelivery deliveryOf(ConsumerRecord record, + List> batch, Acknowledgment springAck) { + return new WireMqDelivery(record.value(), headersOf(record), false, + new KafkaAcknowledgment(springAck, batch)); + } + + private Map headersOf(ConsumerRecord record) { + Map headers = new HashMap<>(); + for (Header header : record.headers()) { + headers.put(header.key(), Objects.isNull(header.value()) ? null + : new String(header.value(), StandardCharsets.UTF_8)); + } + return headers; + } + + private ProducerRecord producerRecord(WireMqMessage wire) { + ProducerRecord record = new ProducerRecord<>(topicName(wire.topic()), + wire.partitionKey(), wire.body()); + wire.headers().forEach((key, value) -> record.headers().add(key, + Objects.isNull(value) ? null : value.getBytes(StandardCharsets.UTF_8))); + return record; + } + + private void publishDead(ConsumerRecord record) { + ProducerRecord dead = new ProducerRecord<>(deadLetterTopic(record.topic()), + record.key(), record.value()); + for (Header header : record.headers()) { + dead.headers().add(header); + } + kafkaTemplate.send(dead); + } + + private static String deadLetterTopic(String topic) { + return topic + ".dlq"; + } + + /** + * Physical topic for a logical destination; logical dead topics map to the + * {@code .dlq} form so rejects and dead-letter subscriptions land on the same topic. + */ + public static String topicName(MqTopic topic) { + return switch (topic) { + case POINT_VALUE_DEAD -> TOPIC_PREFIX + "point_value.dlq"; + case POINT_COMMAND_DEAD -> TOPIC_PREFIX + "point_command.dlq"; + case COMMAND_DEAD -> TOPIC_PREFIX + "command.dlq"; + default -> TOPIC_PREFIX + topic.name().toLowerCase(); + }; + } + + /** + * Port acknowledgment over spring-kafka's handle: ack commits the offset(s), + * reject(true) nacks for near-immediate redelivery, reject(false) republishes the + * record(s) to the dead-letter topic and commits. + */ + private final class KafkaAcknowledgment implements io.github.pnoker.common.mq.listener.Acknowledgment { + + private final Acknowledgment springAcknowledgment; + private final List> records; + + private KafkaAcknowledgment(Acknowledgment springAcknowledgment, + List> records) { + this.springAcknowledgment = springAcknowledgment; + this.records = records; + } + + @Override + public void ack() { + springAcknowledgment.acknowledge(); + } + + @Override + public void reject(boolean requeue) { + if (requeue) { + springAcknowledgment.nack(Duration.ofMillis(50)); + return; + } + records.forEach(KafkaMqAdapter.this::publishDead); + springAcknowledgment.acknowledge(); + } + } +} diff --git a/dc3-common/dc3-common-mq-kafka/src/main/java/io/github/pnoker/common/mq/kafka/config/KafkaMqAdapterConfiguration.java b/dc3-common/dc3-common-mq-kafka/src/main/java/io/github/pnoker/common/mq/kafka/config/KafkaMqAdapterConfiguration.java new file mode 100644 index 000000000..432e05d81 --- /dev/null +++ b/dc3-common/dc3-common-mq-kafka/src/main/java/io/github/pnoker/common/mq/kafka/config/KafkaMqAdapterConfiguration.java @@ -0,0 +1,56 @@ +/* + * Copyright 2016-present the IoT DC3 original author or authors. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +package io.github.pnoker.common.mq.kafka.config; + +import io.github.pnoker.common.mq.config.BatchConsumerProperties; +import io.github.pnoker.common.mq.kafka.KafkaMqAdapter; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.autoconfigure.AutoConfiguration; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.context.annotation.Bean; +import org.springframework.kafka.core.KafkaTemplate; + +import java.util.Map; + +/** + * Activates the Kafka adapter when {@code dc3.mq.type=kafka}. Bootstrap servers come + * from {@code spring.kafka.bootstrap-servers} (standard spring-kafka configuration). + * + * @author pnoker + * @since 2026.8.19 + */ +@AutoConfiguration +@ConditionalOnProperty(prefix = "dc3.mq", name = "type", havingValue = "kafka") +public class KafkaMqAdapterConfiguration { + + @Bean + @ConditionalOnMissingBean(KafkaTemplate.class) + public KafkaTemplate kafkaMqTemplate( + @Value("${spring.kafka.bootstrap-servers:localhost:9092}") String bootstrapServers) { + return KafkaMqAdapter.template(bootstrapServers); + } + + @Bean + public KafkaMqAdapter kafkaMqAdapter(KafkaTemplate kafkaTemplate, + @Value("${spring.kafka.bootstrap-servers:localhost:9092}") + String bootstrapServers, BatchConsumerProperties batchProperties) { + Map consumerConfig = KafkaMqAdapter.consumerConfig(bootstrapServers); + return new KafkaMqAdapter(kafkaTemplate, consumerConfig, batchProperties); + } +} diff --git a/dc3-common/dc3-common-mq-kafka/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports b/dc3-common/dc3-common-mq-kafka/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports new file mode 100644 index 000000000..8796fd365 --- /dev/null +++ b/dc3-common/dc3-common-mq-kafka/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports @@ -0,0 +1,18 @@ +# +# Copyright 2016-present the IoT DC3 original author or authors. +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as +# published by the Free Software Foundation, either version 3 of the +# License, or (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . +# + +io.github.pnoker.common.mq.kafka.config.KafkaMqAdapterConfiguration diff --git a/dc3-common/dc3-common-mq-rabbitmq/src/main/java/io/github/pnoker/common/mq/rabbit/RabbitMqAdapter.java b/dc3-common/dc3-common-mq-rabbitmq/src/main/java/io/github/pnoker/common/mq/rabbit/RabbitMqAdapter.java index db83900ad..484df8c87 100644 --- a/dc3-common/dc3-common-mq-rabbitmq/src/main/java/io/github/pnoker/common/mq/rabbit/RabbitMqAdapter.java +++ b/dc3-common/dc3-common-mq-rabbitmq/src/main/java/io/github/pnoker/common/mq/rabbit/RabbitMqAdapter.java @@ -42,6 +42,7 @@ 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.rabbit.listener.api.ChannelAwareBatchMessageListener; import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener; import org.springframework.amqp.rabbit.retry.MessageBatchRecoverer; import org.aopalliance.aop.Advice; @@ -98,7 +99,10 @@ public class RabbitMqAdapter implements BrokerAdapter { @Override public BrokerCapabilities capabilities() { - return new BrokerCapabilities(true, true, true, true, true, true, true, OrderingGuarantee.NONE); + // delayedMessage=false: only the intrinsic TTL+DLX topics (STATE_TIMEOUT, + // DEVICE_SCAN) delay server-side, and their senders do not set message delays. + // Arbitrary per-message delays go through the port's local-scheduler fallback. + return new BrokerCapabilities(false, true, true, true, true, true, true, OrderingGuarantee.NONE); } @Override @@ -148,17 +152,8 @@ public class RabbitMqAdapter implements BrokerAdapter { container.setBatchSize(batchProperties.getBatchSize()); container.setBatchReceiveTimeout(batchProperties.getReceiveTimeoutMillis()); container.setAdviceChain(batchRetryAdvice()); - container.setMessageListener(new ChannelAwareMessageListener() { - @Override - public void onMessage(Message message, Channel channel) { - handleBatch(List.of(message), channel, listener); - } - - @Override - public void onMessageBatch(List messages, Channel channel) { - handleBatch(messages, channel, listener); - } - }); + container.setMessageListener((ChannelAwareBatchMessageListener) (messages, channel) + -> handleBatch(messages, channel, listener)); start(spec, queue, container); } @@ -298,20 +293,32 @@ public class RabbitMqAdapter implements BrokerAdapter { }; } + + /** + * Blank group resolves to the platform-shared queue (pre-port name); a named group + * gets a group-suffixed copy of the same queue for named consumer groups. + */ + private String grouped(String baseQueue, String group) { + if (Objects.isNull(group) || group.isBlank()) { + return baseQueue; + } + return RabbitTopology.declareGroupedQueue(rabbitAdmin, baseQueue, group); + } + private String resolveQueue(SubscriptionSpec spec) { String group = spec.group(); String keyPattern = spec.keyPattern(); return switch (spec.topic()) { case STATE -> switch (keyPattern) { - case "driver.*" -> RabbitNames.QUEUE_DRIVER_STATE; - case "device.*" -> RabbitNames.QUEUE_DEVICE_STATE; + case "driver.*" -> grouped(RabbitNames.QUEUE_DRIVER_STATE, group); + case "device.*" -> grouped(RabbitNames.QUEUE_DEVICE_STATE, group); default -> throw new IllegalArgumentException( "STATE subscription requires keyPattern driver.* or device.*, got: " + keyPattern); }; case ALARM -> switch (keyPattern) { - case "driver.*" -> RabbitNames.QUEUE_DRIVER_ALARM; - case "device.*" -> RabbitNames.QUEUE_DEVICE_ALARM; - case "task.*" -> RabbitNames.QUEUE_NOTIFY_TASK; + case "driver.*" -> grouped(RabbitNames.QUEUE_DRIVER_ALARM, group); + case "device.*" -> grouped(RabbitNames.QUEUE_DEVICE_ALARM, group); + case "task.*" -> grouped(RabbitNames.QUEUE_NOTIFY_TASK, group); default -> throw new IllegalArgumentException( "ALARM subscription requires keyPattern driver.*, device.* or task.*, got: " + keyPattern); }; @@ -320,8 +327,8 @@ public class RabbitMqAdapter implements BrokerAdapter { RabbitNames.ROUTING_DRIVER_METADATA_PREFIX + keyPattern); yield RabbitNames.QUEUE_DRIVER_METADATA_PREFIX + group; } - case POINT_VALUE -> RabbitNames.QUEUE_POINT_VALUE; - case EVENT -> RabbitNames.QUEUE_EVENT_REPORT; + case POINT_VALUE -> grouped(RabbitNames.QUEUE_POINT_VALUE, group); + case EVENT -> grouped(RabbitNames.QUEUE_EVENT_REPORT, group); case COMMAND -> { RabbitTopology.declareDriverCommandQueue(rabbitAdmin, RabbitNames.QUEUE_COMMAND_PREFIX + group, RabbitNames.EXCHANGE_COMMAND, RabbitNames.EXCHANGE_COMMAND_DEAD, @@ -329,20 +336,22 @@ public class RabbitMqAdapter implements BrokerAdapter { yield RabbitNames.QUEUE_COMMAND_PREFIX + group; } case POINT_COMMAND -> { + int expires = Objects.nonNull(spec.instanceTtl()) && !spec.instanceTtl().isZero() + ? (int) Math.min(spec.instanceTtl().toMillis(), Integer.MAX_VALUE) + : driverQueueExpiresMillis; RabbitTopology.declareDriverCommandQueue(rabbitAdmin, RabbitNames.QUEUE_POINT_COMMAND_PREFIX + group, RabbitNames.EXCHANGE_POINT_COMMAND, RabbitNames.EXCHANGE_POINT_COMMAND_DEAD, - RabbitNames.ROUTING_POINT_COMMAND_PREFIX + keyPattern, driverQueueExpiresMillis); + RabbitNames.ROUTING_POINT_COMMAND_PREFIX + keyPattern, expires); yield RabbitNames.QUEUE_POINT_COMMAND_PREFIX + group; } - case COMMAND_RESULT -> RabbitNames.QUEUE_COMMAND_RESULT; - case POINT_COMMAND_RESULT -> RabbitNames.QUEUE_POINT_COMMAND_RESULT; - case NOTIFY_TASK -> RabbitNames.QUEUE_NOTIFY_TASK; - case STATE_TIMEOUT -> RabbitNames.QUEUE_DRIVER_TIMEOUT_CHECK; - case DEVICE_SCAN -> RabbitNames.QUEUE_DEVICE_SCAN; - case POINT_COMMAND_DEAD -> RabbitNames.QUEUE_POINT_COMMAND_DEAD; - case COMMAND_DEAD -> RabbitNames.QUEUE_COMMAND_DEAD; - case POINT_VALUE_DEAD -> throw new IllegalArgumentException( - "POINT_VALUE_DEAD is a consumer-less quarantine by design"); + case COMMAND_RESULT -> grouped(RabbitNames.QUEUE_COMMAND_RESULT, group); + case POINT_COMMAND_RESULT -> grouped(RabbitNames.QUEUE_POINT_COMMAND_RESULT, group); + case NOTIFY_TASK -> grouped(RabbitNames.QUEUE_NOTIFY_TASK, group); + case STATE_TIMEOUT -> grouped(RabbitNames.QUEUE_DRIVER_TIMEOUT_CHECK, group); + case DEVICE_SCAN -> grouped(RabbitNames.QUEUE_DEVICE_SCAN, group); + case POINT_COMMAND_DEAD -> grouped(RabbitNames.QUEUE_POINT_COMMAND_DEAD, group); + case COMMAND_DEAD -> grouped(RabbitNames.QUEUE_COMMAND_DEAD, group); + case POINT_VALUE_DEAD -> grouped(RabbitNames.QUEUE_POINT_VALUE_DEAD, group); }; } } diff --git a/dc3-common/dc3-common-mq-rabbitmq/src/main/java/io/github/pnoker/common/mq/rabbit/RabbitTopology.java b/dc3-common/dc3-common-mq-rabbitmq/src/main/java/io/github/pnoker/common/mq/rabbit/RabbitTopology.java index 0d7e34582..b68905a4b 100644 --- a/dc3-common/dc3-common-mq-rabbitmq/src/main/java/io/github/pnoker/common/mq/rabbit/RabbitTopology.java +++ b/dc3-common/dc3-common-mq-rabbitmq/src/main/java/io/github/pnoker/common/mq/rabbit/RabbitTopology.java @@ -24,6 +24,11 @@ import org.springframework.amqp.core.QueueBuilder; import org.springframework.amqp.core.TopicExchange; import org.springframework.amqp.rabbit.core.RabbitAdmin; +import java.util.List; +import java.util.Objects; + +import java.util.List; + /** * Declares the physical RabbitMQ topology (exchanges, queues, bindings) with arguments * byte-for-byte identical to the pre-port {@code ExchangeConfig}/{@code DataTopicConfig} @@ -31,6 +36,12 @@ import org.springframework.amqp.rabbit.core.RabbitAdmin; * declarations are idempotent, so a rolling deployment across mixed versions converges * on the same topology. * + *

Platform-shared queues have a descriptor table; a subscription carrying a non-blank + * {@code group} gets a group-suffixed copy of the same queue (same arguments, same + * binding). The blank-group names are exactly the pre-port ones, so production + * deployments are unchanged — the suffix exists for named consumer groups and for + * contract-test isolation. + * * @author pnoker * @since 2026.8.19 */ @@ -41,6 +52,57 @@ public final class RabbitTopology { */ private static final String BINDING_AUTO_DELETE = "x-auto-delete"; + /** + * Descriptor of a platform-shared queue. + * + * @param queueName queue name + * @param exchangeName source exchange + * @param routingKey binding routing key (pattern) + * @param ttlMillis per-queue message TTL, 0 = none + * @param deadExchange dead-letter exchange, null = none + * @param deadRouting dead-letter routing key + * @param bindingArgument whether the binding carries the x-auto-delete argument + */ + private record SharedQueue(String queueName, String exchangeName, String routingKey, int ttlMillis, + String deadExchange, String deadRouting, boolean bindingArgument) { + } + + private static final List SHARED_QUEUES = List.of( + new SharedQueue(RabbitNames.QUEUE_DRIVER_STATE, RabbitNames.EXCHANGE_STATE, + "dc3.r.state.driver.*", 30_000, null, null, true), + new SharedQueue(RabbitNames.QUEUE_DEVICE_STATE, RabbitNames.EXCHANGE_STATE, + "dc3.r.state.device.*", 30_000, null, null, true), + new SharedQueue(RabbitNames.QUEUE_DRIVER_ALARM, RabbitNames.EXCHANGE_ALARM, + "dc3.r.alarm.driver.*", 30_000, null, null, true), + new SharedQueue(RabbitNames.QUEUE_DEVICE_ALARM, RabbitNames.EXCHANGE_ALARM, + "dc3.r.alarm.device.*", 30_000, null, null, true), + new SharedQueue(RabbitNames.QUEUE_POINT_VALUE, RabbitNames.EXCHANGE_VALUE, + "dc3.r.value.point.*", 604_800_000, RabbitNames.EXCHANGE_POINT_VALUE_DEAD, "#", true), + new SharedQueue(RabbitNames.QUEUE_POINT_VALUE_DEAD, RabbitNames.EXCHANGE_POINT_VALUE_DEAD, + "#", 0, null, null, false), + new SharedQueue(RabbitNames.QUEUE_NOTIFY_TASK, RabbitNames.EXCHANGE_ALARM, + "dc3.r.notify.task.*", 86_400_000, null, null, true), + new SharedQueue(RabbitNames.QUEUE_DRIVER_TIMEOUT_DELAY, RabbitNames.EXCHANGE_STATE_TIMEOUT_DELAY, + RabbitNames.ROUTING_DRIVER_TIMEOUT_DELAY, 45_000, + RabbitNames.EXCHANGE_STATE_TIMEOUT_CHECK, RabbitNames.ROUTING_DRIVER_TIMEOUT_CHECK, false), + new SharedQueue(RabbitNames.QUEUE_DRIVER_TIMEOUT_CHECK, RabbitNames.EXCHANGE_STATE_TIMEOUT_CHECK, + RabbitNames.ROUTING_DRIVER_TIMEOUT_CHECK, 0, null, null, false), + new SharedQueue(RabbitNames.QUEUE_DEVICE_SCAN_TICK, RabbitNames.EXCHANGE_STATE_TIMEOUT_DELAY, + RabbitNames.ROUTING_DEVICE_SCAN_TICK, 10_000, + RabbitNames.EXCHANGE_STATE_TIMEOUT_CHECK, RabbitNames.ROUTING_DEVICE_SCAN, false), + new SharedQueue(RabbitNames.QUEUE_DEVICE_SCAN, RabbitNames.EXCHANGE_STATE_TIMEOUT_CHECK, + RabbitNames.ROUTING_DEVICE_SCAN, 0, null, null, false), + new SharedQueue(RabbitNames.QUEUE_POINT_COMMAND_DEAD, RabbitNames.EXCHANGE_POINT_COMMAND_DEAD, + "#", 0, null, null, false), + new SharedQueue(RabbitNames.QUEUE_COMMAND_DEAD, RabbitNames.EXCHANGE_COMMAND_DEAD, + "#", 0, null, null, false), + new SharedQueue(RabbitNames.QUEUE_POINT_COMMAND_RESULT, RabbitNames.EXCHANGE_POINT_COMMAND_RESULT, + "dc3.r.point_command_result.*", 60_000, null, null, false), + new SharedQueue(RabbitNames.QUEUE_COMMAND_RESULT, RabbitNames.EXCHANGE_COMMAND_RESULT, + RabbitNames.ROUTING_COMMAND_RESULT_PREFIX + "*", 60_000, null, null, false), + new SharedQueue(RabbitNames.QUEUE_EVENT_REPORT, RabbitNames.EXCHANGE_EVENT, + RabbitNames.ROUTING_EVENT_PREFIX + "*", 60_000, null, null, false)); + private RabbitTopology() { throw new IllegalStateException("Utility class"); } @@ -52,78 +114,34 @@ public final class RabbitTopology { * @param admin the rabbit admin to declare through */ public static void declareSharedTopology(RabbitAdmin admin) { - TopicExchange state = declareExchange(admin, RabbitNames.EXCHANGE_STATE); - TopicExchange alarm = declareExchange(admin, RabbitNames.EXCHANGE_ALARM); - TopicExchange metadata = declareExchange(admin, RabbitNames.EXCHANGE_METADATA); - TopicExchange pointCommand = declareExchange(admin, RabbitNames.EXCHANGE_POINT_COMMAND); - TopicExchange value = declareExchange(admin, RabbitNames.EXCHANGE_VALUE); - TopicExchange timeoutDelay = declareExchange(admin, RabbitNames.EXCHANGE_STATE_TIMEOUT_DELAY); - TopicExchange timeoutCheck = declareExchange(admin, RabbitNames.EXCHANGE_STATE_TIMEOUT_CHECK); - TopicExchange command = declareExchange(admin, RabbitNames.EXCHANGE_COMMAND); - TopicExchange commandResult = declareExchange(admin, RabbitNames.EXCHANGE_COMMAND_RESULT); - TopicExchange commandDead = declareExchange(admin, RabbitNames.EXCHANGE_COMMAND_DEAD); - TopicExchange event = declareExchange(admin, RabbitNames.EXCHANGE_EVENT); - TopicExchange pointValueDead = declareExchange(admin, RabbitNames.EXCHANGE_POINT_VALUE_DEAD); - TopicExchange pointCommandDead = declareExchange(admin, RabbitNames.EXCHANGE_POINT_COMMAND_DEAD); - declareExchange(admin, RabbitNames.EXCHANGE_POINT_COMMAND_RESULT); + for (String exchange : new String[]{ + RabbitNames.EXCHANGE_STATE, RabbitNames.EXCHANGE_ALARM, RabbitNames.EXCHANGE_METADATA, + RabbitNames.EXCHANGE_POINT_COMMAND, RabbitNames.EXCHANGE_VALUE, + RabbitNames.EXCHANGE_STATE_TIMEOUT_DELAY, RabbitNames.EXCHANGE_STATE_TIMEOUT_CHECK, + RabbitNames.EXCHANGE_COMMAND, RabbitNames.EXCHANGE_COMMAND_RESULT, + RabbitNames.EXCHANGE_COMMAND_DEAD, RabbitNames.EXCHANGE_EVENT, + RabbitNames.EXCHANGE_POINT_VALUE_DEAD, RabbitNames.EXCHANGE_POINT_COMMAND_DEAD, + RabbitNames.EXCHANGE_POINT_COMMAND_RESULT}) { + admin.declareExchange(new TopicExchange(exchange, true, false)); + } + SHARED_QUEUES.forEach(def -> declare(admin, def, def.queueName())); + } - // state / alarm queues, 30 s TTL (these bindings carry the x-auto-delete argument) - bind(admin, QueueBuilder.durable(RabbitNames.QUEUE_DRIVER_STATE).ttl(30_000).build(), - state, "dc3.r.state.driver.*"); - bind(admin, QueueBuilder.durable(RabbitNames.QUEUE_DEVICE_STATE).ttl(30_000).build(), - state, "dc3.r.state.device.*"); - bind(admin, QueueBuilder.durable(RabbitNames.QUEUE_DRIVER_ALARM).ttl(30_000).build(), - alarm, "dc3.r.alarm.driver.*"); - bind(admin, QueueBuilder.durable(RabbitNames.QUEUE_DEVICE_ALARM).ttl(30_000).build(), - alarm, "dc3.r.alarm.device.*"); - - // point value: 7 d TTL then dead-letter to the quarantine exchange - bind(admin, QueueBuilder.durable(RabbitNames.QUEUE_POINT_VALUE) - .ttl(604_800_000) - .deadLetterExchange(RabbitNames.EXCHANGE_POINT_VALUE_DEAD) - .deadLetterRoutingKey("#") - .build(), - value, "dc3.r.value.point.*"); - bindPlain(admin, QueueBuilder.durable(RabbitNames.QUEUE_POINT_VALUE_DEAD).build(), pointValueDead, "#"); - - // notify task: 24 h TTL guard against runaway outbound backlog - bind(admin, QueueBuilder.durable(RabbitNames.QUEUE_NOTIFY_TASK).ttl(86_400_000).build(), - alarm, "dc3.r.notify.task.*"); - - // driver timeout delay chain (45 s TTL + DLX) - bindPlain(admin, QueueBuilder.durable(RabbitNames.QUEUE_DRIVER_TIMEOUT_DELAY) - .ttl(45_000) - .deadLetterExchange(RabbitNames.EXCHANGE_STATE_TIMEOUT_CHECK) - .deadLetterRoutingKey(RabbitNames.ROUTING_DRIVER_TIMEOUT_CHECK) - .build(), - timeoutDelay, RabbitNames.ROUTING_DRIVER_TIMEOUT_DELAY); - bindPlain(admin, QueueBuilder.durable(RabbitNames.QUEUE_DRIVER_TIMEOUT_CHECK).build(), - timeoutCheck, RabbitNames.ROUTING_DRIVER_TIMEOUT_CHECK); - - // device scan tick chain (10 s TTL + DLX) - bindPlain(admin, QueueBuilder.durable(RabbitNames.QUEUE_DEVICE_SCAN_TICK) - .ttl(10_000) - .deadLetterExchange(RabbitNames.EXCHANGE_STATE_TIMEOUT_CHECK) - .deadLetterRoutingKey(RabbitNames.ROUTING_DEVICE_SCAN) - .build(), - timeoutDelay, RabbitNames.ROUTING_DEVICE_SCAN_TICK); - bindPlain(admin, QueueBuilder.durable(RabbitNames.QUEUE_DEVICE_SCAN).build(), - timeoutCheck, RabbitNames.ROUTING_DEVICE_SCAN); - - // dead letters and results (plain bindings, no binding argument) - bindPlain(admin, QueueBuilder.durable(RabbitNames.QUEUE_POINT_COMMAND_DEAD).build(), pointCommandDead, "#"); - bindPlain(admin, QueueBuilder.durable(RabbitNames.QUEUE_COMMAND_DEAD).build(), commandDead, "#"); - bindPlain(admin, QueueBuilder.durable(RabbitNames.QUEUE_POINT_COMMAND_RESULT).ttl(60_000).build(), - declareExchange(admin, RabbitNames.EXCHANGE_POINT_COMMAND_RESULT), - "dc3.r.point_command_result.*"); - bindPlain(admin, QueueBuilder.durable(RabbitNames.QUEUE_COMMAND_RESULT).ttl(60_000).build(), - commandResult, RabbitNames.ROUTING_COMMAND_RESULT_PREFIX + "*"); - bindPlain(admin, QueueBuilder.durable(RabbitNames.QUEUE_EVENT_REPORT).ttl(60_000).build(), - event, RabbitNames.ROUTING_EVENT_PREFIX + "*"); - - // metadata / point command / command exchanges are also consumed by per-instance - // driver queues declared on subscribe; the exchanges above are already declared. - assert metadata != null && pointCommand != null && command != null; + /** + * Declare a group-suffixed copy of a platform-shared queue (same arguments, same + * binding) for named consumer groups; used by the contract suite for isolation. + * + * @param admin rabbit admin + * @param baseQueue base queue name from {@link RabbitNames} + * @param group consumer group suffix + * @return the suffixed queue name + */ + public static String declareGroupedQueue(RabbitAdmin admin, String baseQueue, String group) { + SHARED_QUEUES.stream() + .filter(def -> def.queueName().equals(baseQueue)) + .findFirst() + .ifPresent(def -> declare(admin, def, baseQueue + "." + group)); + return baseQueue + "." + group; } /** @@ -173,21 +191,22 @@ public final class RabbitTopology { admin.declareBinding(binding); } - private static TopicExchange declareExchange(RabbitAdmin admin, String name) { - TopicExchange exchange = new TopicExchange(name, true, false); - admin.declareExchange(exchange); - return exchange; - } - - private static void bind(RabbitAdmin admin, Queue queue, TopicExchange exchange, String routingKey) { - Binding binding = BindingBuilder.bind(queue).to(exchange).with(routingKey); - binding.addArgument(BINDING_AUTO_DELETE, false); + private static void declare(RabbitAdmin admin, SharedQueue def, String queueName) { + QueueBuilder builder = QueueBuilder.durable(queueName); + if (def.ttlMillis() > 0) { + builder.ttl(def.ttlMillis()); + } + if (Objects.nonNull(def.deadExchange())) { + builder.deadLetterExchange(def.deadExchange()).deadLetterRoutingKey(def.deadRouting()); + } + Queue queue = builder.build(); + Binding binding = BindingBuilder.bind(queue) + .to(new TopicExchange(def.exchangeName())) + .with(def.routingKey()); + if (def.bindingArgument()) { + binding.addArgument(BINDING_AUTO_DELETE, false); + } admin.declareQueue(queue); admin.declareBinding(binding); } - - private static void bindPlain(RabbitAdmin admin, Queue queue, TopicExchange exchange, String routingKey) { - admin.declareQueue(queue); - admin.declareBinding(BindingBuilder.bind(queue).to(exchange).with(routingKey)); - } } diff --git a/dc3-common/dc3-common-mq-tck/pom.xml b/dc3-common/dc3-common-mq-tck/pom.xml new file mode 100644 index 000000000..8f7e89347 --- /dev/null +++ b/dc3-common/dc3-common-mq-tck/pom.xml @@ -0,0 +1,96 @@ + + + + 4.0.0 + + + io.github.pnoker + dc3-common + 2026.5.22 + + + ${project.artifactId} + dc3-common-mq-tck + 2026.5.22 + jar + + IoT DC3 broker-neutral contract suite: an adapter that passes these tests is compliant + + + + io.github.pnoker + dc3-common-mq + + + org.junit.jupiter + junit-jupiter-api + + + org.awaitility + awaitility + + + org.assertj + assertj-core + + + org.slf4j + slf4j-api + + + + + io.github.pnoker + dc3-common-mq-rabbitmq + test + + + io.github.pnoker + dc3-common-mq-kafka + test + + + org.junit.jupiter + junit-jupiter + test + + + org.testcontainers + testcontainers-junit-jupiter + test + + + org.testcontainers + testcontainers-rabbitmq + test + + + org.testcontainers + testcontainers-kafka + test + + + org.junit.jupiter + junit-jupiter-engine + test + + + + diff --git a/dc3-common/dc3-common-mq-tck/src/main/java/io/github/pnoker/common/mq/tck/AbstractMqContractTest.java b/dc3-common/dc3-common-mq-tck/src/main/java/io/github/pnoker/common/mq/tck/AbstractMqContractTest.java new file mode 100644 index 000000000..3eed9d9d4 --- /dev/null +++ b/dc3-common/dc3-common-mq-tck/src/main/java/io/github/pnoker/common/mq/tck/AbstractMqContractTest.java @@ -0,0 +1,353 @@ +/* + * Copyright 2016-present the IoT DC3 original author or authors. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +package io.github.pnoker.common.mq.tck; + +import io.github.pnoker.common.constant.mq.ConsumptionProfile; +import io.github.pnoker.common.constant.mq.DeliveryMode; +import io.github.pnoker.common.constant.mq.MqTopic; +import io.github.pnoker.common.constant.mq.SubscriptionMode; +import io.github.pnoker.common.mq.MqHeaders; +import io.github.pnoker.common.mq.adapter.BrokerAdapter; +import io.github.pnoker.common.mq.adapter.RawBatchListener; +import io.github.pnoker.common.mq.adapter.RawDeliveryListener; +import io.github.pnoker.common.mq.adapter.WireMqDelivery; +import io.github.pnoker.common.mq.core.EnvelopeCodec; +import io.github.pnoker.common.mq.core.MessageSenderImpl; +import io.github.pnoker.common.mq.message.MqMessage; +import io.github.pnoker.common.mq.sender.MessageSender; +import io.github.pnoker.common.mq.subscription.SubscriptionSpec; +import org.awaitility.Awaitility; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.slf4j.MDC; + +import java.time.Duration; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.Objects; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Consumer; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Broker-neutral contract suite (docs/design/mq-abstraction.md §11). A broker adapter + * passes this suite ⇒ it is compliant. Community adapters depend on this module and + * provide a harness that instantiates their {@link BrokerAdapter} against a live broker. + * Cases use unique groups and payload markers so a fresh broker per run keeps them + * isolated. + * + * @author pnoker + * @since 2026.8.19 + */ +public abstract class AbstractMqContractTest { + + /** + * Simple wire payload with a unique marker per message. + */ + public record TestPayload(String id, String text) { + } + + /** + * Collector pairing received payloads with their headers. + */ + protected static final class Received { + final TestPayload payload; + final Map headers; + final boolean redelivered; + + Received(WireMqDelivery delivery) { + this.payload = EnvelopeCodec.deserialize(delivery, TestPayload.class); + this.headers = delivery.headers(); + this.redelivered = delivery.redelivered(); + } + } + + /** + * Unique suffix per suite run so repeated runs against a long-lived broker never + * observe each other's messages. + */ + protected static final String RUN = UUID.randomUUID().toString().substring(0, 8); + + protected abstract BrokerAdapter adapter(); + + protected void shutdownAdapter() { + } + + @AfterEach + void tearDown() { + shutdownAdapter(); + MDC.remove(io.github.pnoker.common.constant.common.RequestIdConstant.MDC_KEY); + } + + protected final MessageSender sender() { + return new MessageSenderImpl(adapter()); + } + + protected final SubscriptionSpec loadBalance(MqTopic topic, String group) { + return new SubscriptionSpec(topic, SubscriptionMode.LOAD_BALANCE, ConsumptionProfile.LATENCY, + DeliveryMode.SINGLE, "", group + "-" + RUN, null, TestPayload.class, true); + } + + protected final SubscriptionSpec loadBalancePattern(MqTopic topic, String group, String keyPattern) { + return loadBalancePattern(topic, group, keyPattern, null); + } + + protected final SubscriptionSpec loadBalancePattern(MqTopic topic, String group, String keyPattern, + java.time.Duration instanceTtl) { + return new SubscriptionSpec(topic, SubscriptionMode.LOAD_BALANCE, ConsumptionProfile.LATENCY, + DeliveryMode.SINGLE, keyPattern, group + "-" + RUN, instanceTtl, TestPayload.class, true); + } + + protected final SubscriptionSpec broadcast(MqTopic topic, String group) { + return new SubscriptionSpec(topic, SubscriptionMode.BROADCAST, ConsumptionProfile.LATENCY, + DeliveryMode.SINGLE, "tckbroadcast", group + "-" + RUN, null, TestPayload.class, true); + } + + protected final SubscriptionSpec batchSpec(MqTopic topic) { + return new SubscriptionSpec(topic, SubscriptionMode.LOAD_BALANCE, ConsumptionProfile.THROUGHPUT, + DeliveryMode.BATCH, "", "tck-batch-" + RUN, null, TestPayload.class, true); + } + + protected final List subscribeCollector(SubscriptionSpec spec, Consumer each) { + List received = new CopyOnWriteArrayList<>(); + adapter().subscribe(spec, delivery -> { + if (Objects.nonNull(each)) { + each.accept(delivery); + } else { + received.add(new Received(delivery)); + delivery.acknowledgment().ack(); + } + }); + settle(); + return received; + } + + protected final List subscribeBatchCollector(SubscriptionSpec spec, + java.util.function.BiConsumer, + io.github.pnoker.common.mq.listener.Acknowledgment> batch) { + List received = new CopyOnWriteArrayList<>(); + adapter().subscribeBatch(spec, deliveries -> { + List batchReceived = deliveries.stream().map(Received::new).toList(); + received.addAll(batchReceived); + if (Objects.nonNull(batch)) { + batch.accept(batchReceived, deliveries.get(0).acknowledgment()); + } else { + deliveries.get(0).acknowledgment().ack(); + } + }); + settle(); + return received; + } + + /** + * Consumer-group join latency allowance: log-based brokers (kafka) with + * auto.offset.reset=latest only see messages published after the group has joined. + */ + protected final void settle() { + try { + Thread.sleep(600); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + + protected final TestPayload payload(String text) { + return new TestPayload(UUID.randomUUID().toString(), text); + } + + protected final void await(String description, java.util.function.BooleanSupplier condition) { + Awaitility.await(description).atMost(Duration.ofSeconds(15)).pollInterval(Duration.ofMillis(100)) + .until(condition::getAsBoolean); + } + + @Test + void roundTripPreservesEnvelopeHeadersAndPayload() { + List received = subscribeCollector(loadBalance(MqTopic.EVENT, "tck-rt"), null); + TestPayload sent = payload("round-trip"); + + MDC.put(io.github.pnoker.common.constant.common.RequestIdConstant.MDC_KEY, "tck-request-1"); + sender().send(MqMessage.of(MqTopic.EVENT, "tckrt", sent)); + + await("round-trip delivery", () -> !received.isEmpty()); + assertThat(received.get(0).payload).isEqualTo(sent); + assertThat(received.get(0).headers).containsEntry(MqHeaders.DC3_TYPE, TestPayload.class.getName()); + assertThat(received.get(0).headers).containsEntry(MqHeaders.REQUEST_ID, "tck-request-1"); + } + + @Test + void loadBalanceDeliversEachMessageExactlyOnceAcrossInstances() { + List first = subscribeCollector(loadBalance(MqTopic.EVENT, "tck-lb"), null); + List second = subscribeCollector(loadBalance(MqTopic.EVENT, "tck-lb"), null); + + MessageSender sender = sender(); + for (int i = 0; i < 6; i++) { + sender.send(MqMessage.of(MqTopic.EVENT, "tcklb", payload("lb-" + i))); + } + + await("all six messages consumed", () -> first.size() + second.size() >= 6); + Set ids = ConcurrentHashMap.newKeySet(); + first.forEach(r -> ids.add(r.payload.id())); + second.forEach(r -> ids.add(r.payload.id())); + assertThat(ids).hasSize(6); + } + + @Test + void broadcastDeliversToEveryInstance() { + List first = subscribeCollector(broadcast(MqTopic.METADATA, "tck-bc-a"), null); + List second = subscribeCollector(broadcast(MqTopic.METADATA, "tck-bc-b"), null); + + MessageSender sender = sender(); + for (int i = 0; i < 3; i++) { + sender.send(MqMessage.of(MqTopic.METADATA, "tckbroadcast", payload("bc-" + i))); + } + + await("first instance saw all", () -> first.size() >= 3); + await("second instance saw all", () -> second.size() >= 3); + } + + @Test + void delayIsRespectedThroughTheFallback() { + List received = subscribeCollector(loadBalance(MqTopic.EVENT, "tck-delay"), null); + + long start = System.nanoTime(); + sender().send(MqMessage.builder() + .topic(MqTopic.EVENT) + .partitionKey("tckdelay") + .payload(payload("delayed")) + .delay(Duration.ofSeconds(2)) + .build()); + + assertThat(received).isEmpty(); + await("delayed delivery", () -> !received.isEmpty()); + long elapsedMillis = (System.nanoTime() - start) / 1_000_000; + assertThat(elapsedMillis).isGreaterThanOrEqualTo(1500); + } + + @Test + void rejectWithoutRequeueRoutesToTheDeadLetter() { + List dead = subscribeCollector(loadBalance(MqTopic.POINT_COMMAND_DEAD, "tck-dlq-reader"), null); + subscribeCollector(loadBalancePattern(MqTopic.POINT_COMMAND, "tck-dlq", "tck.*"), + delivery -> delivery.acknowledgment().reject(false)); + + TestPayload sent = payload("doomed"); + sender().send(MqMessage.of(MqTopic.POINT_COMMAND, "tck.node", sent)); + + await("message reaches the dead letter", () -> dead.stream().anyMatch(r -> r.payload.equals(sent))); + } + + @Test + void rejectWithRequeueRedelivers() { + AtomicInteger attempts = new AtomicInteger(); + subscribeCollector(loadBalancePattern(MqTopic.POINT_COMMAND, "tck-rq", "tck.*"), delivery -> { + if (attempts.getAndIncrement() == 0) { + delivery.acknowledgment().reject(true); + } else { + delivery.acknowledgment().ack(); + } + }); + + sender().send(MqMessage.of(MqTopic.POINT_COMMAND, "tck.node", payload("retry"))); + + await("redelivery observed", () -> attempts.get() >= 2); + } + + @Test + void sendAsyncConfirmationFires() throws Exception { + subscribeCollector(loadBalance(MqTopic.EVENT, "tck-confirm"), null); + CountDownLatch confirmed = new CountDownLatch(1); + AtomicInteger outcome = new AtomicInteger(-1); + + sender().sendAsync(MqMessage.of(MqTopic.EVENT, "tckconfirm", payload("confirmed")), + (message, ok, cause) -> { + outcome.set(ok ? 1 : 0); + confirmed.countDown(); + }); + + assertThat(confirmed.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(outcome.get()).isEqualTo(1); + } + + @Test + void burstOfMessagesIsNotLost() { + List received = subscribeCollector(loadBalance(MqTopic.EVENT, "tck-burst"), null); + + MessageSender sender = sender(); + for (int i = 0; i < 100; i++) { + sender.send(MqMessage.of(MqTopic.EVENT, "tckburst", payload("burst-" + i))); + } + + await("all 100 burst messages delivered", () -> received.size() >= 100); + } + + @Test + void batchDeliveryCommitsTheWholeBatch() { + List received = subscribeBatchCollector(batchSpec(MqTopic.POINT_VALUE), null); + + MessageSender sender = sender(); + for (int i = 0; i < 10; i++) { + sender.send(MqMessage.of(MqTopic.POINT_VALUE, "tckbatch", payload("batch-" + i))); + } + + await("all batch messages delivered", () -> received.size() >= 10); + int afterAck = received.size(); + Awaitility.await().during(Duration.ofSeconds(1)).atMost(Duration.ofSeconds(2)) + .until(() -> received.size() == afterAck); + } + + @Test + void retryExhaustionDeadLettersInsteadOfDropping() { + List dead = subscribeCollector(loadBalance(MqTopic.POINT_VALUE_DEAD, "tck-dead-reader"), null); + subscribeBatchCollector(batchSpec(MqTopic.POINT_VALUE), (batch, ack) -> { + throw new IllegalStateException("always failing listener"); + }); + + TestPayload sent = payload("exhausted"); + sender().send(MqMessage.of(MqTopic.POINT_VALUE, "tckbatch", sent)); + + await("exhausted retries dead-letter the message", + () -> dead.stream().anyMatch(r -> r.payload.equals(sent))); + } + + @Test + void messagesSurviveWhileNoConsumerIsRunning() { + List first = subscribeCollector(loadBalancePattern(MqTopic.COMMAND, "tck-durable", "tck.*"), null); + shutdownAdapter(); + + MessageSender sender = sender(); + for (int i = 0; i < 3; i++) { + sender.send(MqMessage.of(MqTopic.COMMAND, "tck.node", payload("durable-" + i))); + } + + List second = subscribeCollector(loadBalancePattern(MqTopic.COMMAND, "tck-durable", "tck.*"), null); + await("durable messages redelivered after restart", () -> second.size() >= 3); + } + + /** + * Per-instance subscription expiry is broker-specific; harnesses without the + * capability override this test with a disabled assumption. + */ + @Test + public abstract void perInstanceSubscriptionExpiresAfterInstanceStops(); +} diff --git a/dc3-common/dc3-common-mq-tck/src/test/java/io/github/pnoker/common/mq/tck/KafkaContractTest.java b/dc3-common/dc3-common-mq-tck/src/test/java/io/github/pnoker/common/mq/tck/KafkaContractTest.java new file mode 100644 index 000000000..d18e79e67 --- /dev/null +++ b/dc3-common/dc3-common-mq-tck/src/test/java/io/github/pnoker/common/mq/tck/KafkaContractTest.java @@ -0,0 +1,95 @@ +/* + * Copyright 2016-present the IoT DC3 original author or authors. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +package io.github.pnoker.common.mq.tck; + +import io.github.pnoker.common.mq.adapter.BrokerAdapter; +import io.github.pnoker.common.mq.config.BatchConsumerProperties; +import io.github.pnoker.common.mq.kafka.KafkaMqAdapter; +import org.junit.jupiter.api.Assumptions; +import org.junit.jupiter.api.Test; +import org.testcontainers.junit.jupiter.Testcontainers; +import org.testcontainers.kafka.KafkaContainer; +import org.testcontainers.utility.DockerImageName; + +import java.util.Objects; + +/** + * Kafka harness for the broker-neutral contract suite. By default a disposable KRaft + * container; setting {@code TCK_KAFKA_BOOTSTRAP} points it at an externally managed + * broker (useful where the managed container misbehaves under a given runtime). + * + * @author pnoker + * @since 2026.8.19 + */ +@Testcontainers(disabledWithoutDocker = true) +class KafkaContractTest extends AbstractMqContractTest { + + private static final String EXTERNAL_BOOTSTRAP = System.getenv("TCK_KAFKA_BOOTSTRAP"); + + // started manually (not via the extension) so TCK_KAFKA_BOOTSTRAP fully bypasses it: + // testcontainers 2.0.5 configures apache/kafka 3.9.0 with a nonroutable advertised + // listener the broker rejects, so some runtimes need an externally managed broker. + private static final KafkaContainer KAFKA = new KafkaContainer(DockerImageName.parse("apache/kafka:3.9.0")); + + private static String bootstrap() { + if (Objects.nonNull(EXTERNAL_BOOTSTRAP)) { + return EXTERNAL_BOOTSTRAP; + } + if (!KAFKA.isRunning()) { + KAFKA.start(); + } + return KAFKA.getBootstrapServers(); + } + + private KafkaMqAdapter kafkaAdapter; + + @Override + protected BrokerAdapter adapter() { + if (Objects.isNull(kafkaAdapter)) { + String bootstrap = bootstrap(); + BatchConsumerProperties properties = new BatchConsumerProperties(); + properties.setBatchSize(10); + properties.setMaxRetries(2); + properties.setRetryInitialIntervalMillis(100); + properties.setRetryMultiplier(2); + properties.setRetryMaxIntervalMillis(200); + kafkaAdapter = new KafkaMqAdapter(KafkaMqAdapter.template(bootstrap), + KafkaMqAdapter.consumerConfig(bootstrap), properties); + } + return kafkaAdapter; + } + + @Override + protected void shutdownAdapter() { + if (Objects.nonNull(kafkaAdapter)) { + kafkaAdapter.stop(); + } + } + + /** + * Kafka offsets persist; there is no per-instance subscription expiry (documented + * cleanup policy, capability false). + */ + @Test + @Override + public void perInstanceSubscriptionExpiresAfterInstanceStops() { + adapter(); + Assumptions.assumeTrue(kafkaAdapter.capabilities().subscriptionExpiry(), + "kafka declares subscriptionExpiry=false; documented cleanup policy applies"); + } +} diff --git a/dc3-common/dc3-common-mq-tck/src/test/java/io/github/pnoker/common/mq/tck/RabbitMqContractTest.java b/dc3-common/dc3-common-mq-tck/src/test/java/io/github/pnoker/common/mq/tck/RabbitMqContractTest.java new file mode 100644 index 000000000..80d8011a1 --- /dev/null +++ b/dc3-common/dc3-common-mq-tck/src/test/java/io/github/pnoker/common/mq/tck/RabbitMqContractTest.java @@ -0,0 +1,112 @@ +/* + * Copyright 2016-present the IoT DC3 original author or authors. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +package io.github.pnoker.common.mq.tck; + +import io.github.pnoker.common.constant.mq.MqTopic; +import io.github.pnoker.common.mq.adapter.BrokerAdapter; +import io.github.pnoker.common.mq.config.BatchConsumerProperties; +import io.github.pnoker.common.mq.rabbit.RabbitMqAdapter; +import org.awaitility.Awaitility; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; +import org.springframework.amqp.rabbit.core.RabbitAdmin; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.testcontainers.containers.RabbitMQContainer; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; +import org.testcontainers.utility.DockerImageName; + +import java.time.Duration; +import java.util.Objects; + +/** + * RabbitMQ harness for the broker-neutral contract suite: disposable container, + * publisher confirms and returns enabled, fast batch/retry tuning. + * + * @author pnoker + * @since 2026.8.19 + */ +@Testcontainers(disabledWithoutDocker = true) +class RabbitMqContractTest extends AbstractMqContractTest { + + @Container + private static final RabbitMQContainer RABBIT = + new RabbitMQContainer(DockerImageName.parse("rabbitmq:3.13-management-alpine")); + + private CachingConnectionFactory connectionFactory; + private RabbitMqAdapter rabbitAdapter; + + @Override + protected BrokerAdapter adapter() { + if (Objects.isNull(rabbitAdapter)) { + connectionFactory = new CachingConnectionFactory(java.net.URI.create(RABBIT.getAmqpUrl())); + connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED); + connectionFactory.setPublisherReturns(true); + RabbitTemplate template = new RabbitTemplate(connectionFactory); + template.setMandatory(true); + RabbitAdmin admin = new RabbitAdmin(connectionFactory); + rabbitAdapter = new RabbitMqAdapter(template, admin, connectionFactory, fastProperties(), 300_000); + } + return rabbitAdapter; + } + + private static BatchConsumerProperties fastProperties() { + BatchConsumerProperties properties = new BatchConsumerProperties(); + properties.setBatchSize(10); + properties.setReceiveTimeoutMillis(100); + properties.setConcurrentConsumers(2); + properties.setMaxConcurrentConsumers(4); + properties.setPrefetchCount(10); + properties.setMaxRetries(2); + properties.setRetryInitialIntervalMillis(100); + properties.setRetryMultiplier(2); + properties.setRetryMaxIntervalMillis(200); + return properties; + } + + @Override + protected void shutdownAdapter() { + if (Objects.nonNull(rabbitAdapter)) { + rabbitAdapter.stop(); + } + } + + /** + * The driver command queue expires (x-expires) once its instance stops; after the + * TTL the broker removes the queue entirely. + */ + @Test + @Override + public void perInstanceSubscriptionExpiresAfterInstanceStops() { + adapter(); + Assertions.assertTrue(rabbitAdapter.capabilities().subscriptionExpiry(), + "rabbitmq declares subscriptionExpiry=true"); + + String group = "tck-ttl"; + subscribeCollector(loadBalancePattern(MqTopic.POINT_COMMAND, group, "tck.*", Duration.ofMillis(600)), + delivery -> delivery.acknowledgment().ack()); + shutdownAdapter(); + + RabbitAdmin admin = new RabbitAdmin(connectionFactory); + Awaitility.await("queue expires after instance stop") + .atMost(Duration.ofSeconds(5)) + .pollInterval(Duration.ofMillis(200)) + .until(() -> Objects.isNull(admin.getQueueProperties("dc3.q.point_command." + group))); + } +} diff --git a/dc3-common/pom.xml b/dc3-common/pom.xml index fd1f60875..e3908a18d 100644 --- a/dc3-common/pom.xml +++ b/dc3-common/pom.xml @@ -94,6 +94,8 @@ dc3-common-mqtt dc3-common-mq dc3-common-mq-rabbitmq + dc3-common-mq-kafka + dc3-common-mq-tck dc3-common-postgres dc3-common-public dc3-common-quartz @@ -380,6 +382,16 @@ dc3-common-mq-rabbitmq ${dc3.version} + + io.github.pnoker + dc3-common-mq-kafka + ${dc3.version} + + + io.github.pnoker + dc3-common-mq-tck + ${dc3.version} + io.github.pnoker dc3-common-repository diff --git a/pom.xml b/pom.xml index e6610ddcf..2e30ef3e3 100644 --- a/pom.xml +++ b/pom.xml @@ -766,6 +766,16 @@ dc3-common-mq-rabbitmq ${dc3.version} + + io.github.pnoker + dc3-common-mq-kafka + ${dc3.version} + + + io.github.pnoker + dc3-common-mq-tck + ${dc3.version} + io.github.pnoker dc3-common-repository