Код: Выделить всё
package test
import com.dumpling.WireMockConfig
import com.dumpling.scheduler.DumplingApplication
import com.dumpling.scheduler.application.helpdesk.HelpdeskComponent
import org.apache.kafka.clients.consumer.Consumer
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.clients.consumer.KafkaConsumer
import org.mockito.Mock
import org.mockito.Mockito
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.boot.test.context.SpringBootTest
import org.springframework.boot.test.context.TestConfiguration
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Import
import org.springframework.kafka.core.KafkaTemplate
import org.springframework.kafka.test.EmbeddedKafkaBroker
import org.springframework.kafka.test.context.EmbeddedKafka
import org.springframework.kafka.test.utils.KafkaTestUtils
import org.springframework.test.context.ActiveProfiles
import org.springframework.test.context.TestPropertySource
import spock.lang.Specification
@SpringBootTest(
webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT, classes = DumplingApplication.class)
@ActiveProfiles("test")
@Import(WireMockConfig.class)
@EmbeddedKafka(partitions = 1, topics = [ "my-topic"])
@TestPropertySource(properties = [
'test.outgoingRouting.response={\"workflow\":[{\"outgoingRouting\":[{\"code\":\"test-pilot\",\"cronExpression\":\"0/1 * * * * ?\",\"description\":\"crazy stuff\",\"routingCriteria\":{\"advanced\":false,\"businessWorkflow\":{\"code\":\"test\"},\"inverseSourceSystem\":false},\"routingInformation\":{\"aggregationCriteria\":{\"account\":false,,\"creditDebitIndicator\":false,\"entity\":false,\"instrument\":false,\"internalClassificationType\":false,\"method\":false,\"status\":false,\"valueDate\":false},\"aggregationType\":\"No Aggregation\",\"allowDuplicates\":true,\"channel\":{\"code\":\"File-System\"},\"encoding\":{\"code\":\"UTF-8\"},groupingCriteria\":{\"CEVClassification\":false,\"confidential\":false,\"currency\":false,\"generateNoticeToReceive\":false,\"ICT\":false},\"publishSourceReference\":false,\"signingMethod\":{\"code\":\"AUTHENTICATION\"}}}]}]}'
])
class SchedulerHelpdeskKafkaTest extends Specification {
@Autowired
HelpdeskComponent helpdeskComponent
@Autowired
EmbeddedKafkaBroker broker
@Mock
KafkaTemplate kafkaTemplate
@TestConfiguration
static class MockKafkaConfig {
@Bean
static KafkaTemplate kafkaTemplate() {
return Mockito.mock(KafkaTemplate.class);
}
}
def "Test 1: Should send a message to Kafka topic my-topic"() {
given:
String topic = "my-topic"
and: "a Kafka consumer"
Map consumerProps = KafkaTestUtils.consumerProps(topic, "true", broker)
consumerProps.put("auto.offset.reset", "earliest")
Consumer consumer = new KafkaConsumer(consumerProps)
consumer.subscribe([topic])
when:
helpdeskComponent.dispatchTriggers()
then:
def records = consumer.poll(java.time.Duration.ofSeconds(5))
records.count() == 1
records.iterator().next().topic() == topic
cleanup:
consumer.close()
}
}
< /code>
Я проверил, что вызывается Helpdescomponent, и он имеет следующую логику. < /p>
package com.dumpling.scheduler.application.helpdesk;
import com.dumpling.scheduler.domain.OutgoingRouting;
import com.dumpling.scheduler.domain.ScheduledJobRepository;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.context.annotation.Bean;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.support.CronTrigger;
import org.springframework.stereotype.Component;
import java.util.Collection;
import java.util.HashSet;
import java.util.Set;
@Component
@RequiredArgsConstructor
@EnableScheduling
@Slf4j
public class HelpdeskComponent {
private final Set scheduledRoutings = new HashSet();
@Value("${dumpling.kafka.helpdesk-topic}")
private String topicName;
private final ScheduledJobRepository repository;
private final TaskScheduler scheduler;
private final HelpdeskKafkaProducer helpdeskKafkaProducer;
@Bean
public CommandLineRunner start() {
return args -> dispatchTriggers();
}
public void dispatchTriggers() {
Collection routings = repository.getSchedules().values();
if (routings.isEmpty()) {
return;
}
routings.forEach(this::scheduleTask);
}
private void scheduleTask(OutgoingRouting outgoingRouting) {
if (!scheduledRoutings.contains(outgoingRouting.getCode())) {
scheduler.schedule(() -> helpdeskKafkaProducer.send(topicName, outgoingRouting.getJson()), new CronTrigger(outgoingRouting.getCronExpression()));
log.info(String.format("%s - Scheduling outgoing routing [%s] with cron expression [%s]", "SCH-helpdesk", outgoingRouting.getCode(), outgoingRouting.getCronExpression()));
scheduledRoutings.add(outgoingRouting.getCode());
}
}
}
< /code>
Это класс helpdeskafkaproducer < /p>
package com.dumpling.scheduler.application.helpdesk;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import java.util.concurrent.CompletableFuture;
@Component
@RequiredArgsConstructor
@Slf4j
public class HelpdeskKafkaProducer {
private final KafkaTemplate kafkaTemplate;
public void send(String topic, String message) {
CompletableFuture future = kafkaTemplate.send(topic, message);
future.whenComplete((result, ex) -> {
throw new IllegalArgumentException("Exception occurred during sending message to Kafka topic.", ex);
});
}
}
< /code>
Когда я запускаю свой тест, я получаю следующую stacktrace: < /p>
Caused by: java.util.concurrent.RejectedExecutionException: Task java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask@54b68f1f[Not completed, task = java.util.concurrent.Executors$RunnableAdapter@7d9c7d6d[Wrapped task = DelegatingErrorHandlingRunnable for com.dumpling.scheduler.application.helpdesk.HelpdeskComponent$$Lambda/0x0000007001eab7d0@5fd7b1cb]] rejected from org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler$1@4e00723b[Shutting down, pool size = 1, active threads = 1, queued tasks = 1, completed tasks = 0]
at java.base/java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2081) ~[na:na]
at java.base/java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:841) ~[na:na]
at java.base/java.util.concurrent.ScheduledThreadPoolExecutor.delayedExecute(ScheduledThreadPoolExecutor.java:340) ~[na:na]
at java.base/java.util.concurrent.ScheduledThreadPoolExecutor.schedule(ScheduledThreadPoolExecutor.java:562) ~[na:na]
at org.springframework.scheduling.concurrent.ReschedulingRunnable.schedule(ReschedulingRunnable.java:83) ~[spring-context-6.2.10.jar:6.2.10]
at org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler.schedule(ThreadPoolTaskScheduler.java:415) ~[spring-context-6.2.10.jar:6.2.10]
... 20 common frames omitted
< /code>
Официальный сбой теста: < /p>
Condition not satisfied:
records.count() == 1
| | |
| 0 false
Подробнее здесь: https://stackoverflow.com/questions/797 ... spock-test
