я пытаюсь из теста отправить сообщение в топик, но оно не приходит
@Slf4j
@Component
@RequiredArgsConstructor
@ConditionalOnProperty(name = "messaging.type", havingValue = "kafka")
public class UserEventKafkaConsumer {

    private final UserEventService userEventService;

    private final WsEventKafkaProducer wsEventKafkaProducer;


    @RetryableTopic(include = {Exception.class},
            attempts = "1",
            topicSuffixingStrategy = TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE,
            retryTopicSuffix = "-custom-try",
            dltTopicSuffix = "-dead-t",
            kafkaTemplate = "kafkaTemplate")
    @KafkaListener(
            groupId = "${spring.kafka.consumer.group-id: user-event-notification-group}",
            topics = "${spring.kafka.consumer.topics.events: oila.notification.event}",
            containerFactory = "registeredEventsFactory")
    public void listen(ConsumerRecord<String, String> data, Acknowledgment ack) {
        ack.acknowledge();
        log.info("Received new event registered message = {}", data.value());
        Optional<UserEventRequestDto> opDto = JsonSerialization.optionalFromJson(data.value(), UserEventRequestDto.class);
        if (opDto.isEmpty()){
            log.error("Failed to deserialize event registered message = {}", data.value());
            return;
        }
        if (userEventService.saveNewEvent(opDto.get())) {
            log.info("New event for client = {} saved ", opDto.get().getOcid());
            wsEventKafkaProducer.send(opDto.get().getOcid());
        } else {
            log.error("Failed to save new event for client = {}", opDto.get().getOcid());
        }
    }


@SpringBootTest
@Testcontainers
public class KafkaListenerMessageTest {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    private TestKafkaMessageListener testKafkaMessageListener;

    @Container
    static final PostgreSQLContainer<?> postgreSQLContainer = new PostgreSQLContainer<>("postgres:15-alpine")
            .withDatabaseName("testdb")
            .withUsername("postgres")
            .withPassword("postgres")
            .withStartupTimeoutSeconds(60)
            .waitingFor(Wait.forListeningPort());

    @DynamicPropertySource
    static void configurePg(DynamicPropertyRegistry registry) {
        registry.add("spring.datasource.url", postgreSQLContainer::getJdbcUrl);
        registry.add("spring.datasource.username", postgreSQLContainer::getUsername);
        registry.add("spring.datasource.password", postgreSQLContainer::getPassword);
    }

    @Container
    static final RabbitMQContainer rabbit = new RabbitMQContainer("rabbitmq:3.12.0");

    @DynamicPropertySource
    static void configureProperties(DynamicPropertyRegistry registry) {
        registry.add("spring.rabbitmq.host", rabbit::getHost);
        registry.add("spring.rabbitmq.port", () -> rabbit.getMappedPort(5672));
        registry.add("spring.rabbitmq.username", rabbit::getAdminUsername);
        registry.add("spring.rabbitmq.password", rabbit::getAdminPassword);
    }

    @Container
    public static KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:7.4.0"));

    @DynamicPropertySource
    static void setKafkaProperties(DynamicPropertyRegistry registry) {
        registry.add("spring.kafka.bootstrap-servers", kafka::getBootstrapServers);
    }


    @BeforeEach
    public void setup() {
        testKafkaMessageListener.clear();
    }

    @Test
    @Sql(scripts = "/scripts/data.sql", executionPhase = Sql.ExecutionPhase.BEFORE_TEST_METHOD)
    void testKafkaListenerReceivesMessage() throws Exception {
        String message = "{"
                + "\"referenceId\": 1,"
                + "\"subject\": \"NEWsubject\","
                + "\"message\": \"NEWmessage\","
                + "\"ocid\": \"23\","
                + "\"notificationParams\": {"
                + "  \"eventType\": \"OPERATION\","
                + "  \"eventSubType\": \"CARD_ATTACHED\""
                + "}"
                + "}";

        kafkaTemplate.send("oila.notification.event", message);
        kafkaTemplate.flush();

        Thread.sleep(2000);
        Assertions.assertNotNull(testKafkaMessageListener.getLastNotificationMessage());
    }
}