Spring не может найти созданную таблицу без перезапуска приложения

Question

У меня довольно специфический сценарий, в котором я столкнулся с проблемами:

У меня есть маршрут Camel, который пытается извлечь данные из произвольного числа источников данных и вставить их в центральную базу данных.Если целевой таблицы не существует в центральной базе данных, я постараюсь получить DDL из исходной базы данных.Создание отсутствующих таблиц в центральной базе данных прекрасно работает, но когда в следующий раз маршрут Camel запускается по расписанию, он не может найти таблицу, хотя я вижу ее в своем инструменте базы данных.Если я просто перезапущу экземпляр весенней загрузки, он найдет таблицу, и она будет работать.

Чтобы дать вам лучшее представление о том, как он работает, вот фрагмент маршрута, где он ошибается:

from("direct:ensureTargetTableExists")
    .routeId("ensureTargetTableExists")
    .doTry()
        .toD(String.format("sql:select * from ${in.header.%s} where false?dataSource=%s&outputHeader=%s", QueryHandler.Properties.TARGET_TABLE, DataSourceHandler.Properties.TARGET_DATASOURCE_NAME, "tableExists"))
        .to("direct:handleResultSet")
    .doCatch(BadSqlGrammarException.class)
        .log("entered catch")
        .log(LoggingLevel.INFO, String.format("Could not find table ${in.header.%s} at %s, will attempt to create it by fetching DDL.", QueryHandler.Properties.TARGET_TABLE, DataSourceHandler.Properties.TARGET_DATASOURCE_NAME))
        .to("direct:fetchTableDdlFromSource")
    .endDoTry();

from("direct:fetchTableDdlFromSource")
    .routeId("fetchTableDdlFromSource")
    .streamCaching()
    .bean(DumpHandler.class, "generateDumpCommand")
    .toD(String.format("ssh:{{sync.dump.ssh.user}}@${in.header.%s}?password={{sync.dump.ssh.password}}", DataSourceHandler.Properties.FETCHED_FROM_HOST))
    .to("direct:applyDdlToTargetDatabase");

from("direct:applyDdlToTargetDatabase")
    .routeId("applyDdlToTargetDatabase")
    .transacted("targetTransactionPolicy")
    .toD(String.format("sql:?dataSource=%s&useMessageBodyForSql=true", DataSourceHandler.Properties.TARGET_DATASOURCE_NAME))
    .bean(FileUtil.class, "saveDdl")
    // Nothing else is called after this, since I want the transaction to be completed and commited
    .log(LoggingLevel.INFO, "Table will be populated next scheduled run.");

Я позволил ему выполнить две итерации, и вот что происходит:

entered catch
Could not find table article_flags at target, will attempt to create it by fetching DDL.
Saving DDL to config/sql/ddl/article_flags-20181206_162310.sql
Successfully created table article_flags
Table will be populated next scheduled run.
entered catch
Could not find table article_flags at target, will attempt to create it by fetching DDL.
Failed delivery for (MessageId: ID-overlord-1544109787810-0-12 on ExchangeId: ID-overlord-1544109787810-0-9). Exhausted after delivery attempt: 1 caught: org.springframework.jdbc.BadSqlGrammarException: PreparedStatementCallback; bad SQL grammar []; nested exception is org.postgresql.util.PSQLException: ERROR: relation "article_flags" already exists

Я пытался решить эту проблему с помощью менеджеров транзакций и политик транзакций, которые я добавляю при запуске приложения, но нетпомогло.Как вы можете видеть, я использую политику транзакций targetTransactionPolicy на верблюжьем маршруте выше.

public class ContextConfiguration implements BeanDefinitionRegistryPostProcessor {

    private static final Logger LOG = LoggerFactory.getLogger(ContextConfiguration.class);

    private Map<String, BasicDataSource> datasources;

    public ContextConfiguration(Map<String, BasicDataSource> datasources) {
        this.datasources = datasources;
    }

    @Override
    public void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry registry) throws BeansException {
        for (Entry<String, BasicDataSource> source : datasources.entrySet()) {
            LOG.info("Adding datasource {} targeting {} to Spring registry", source.getKey(), source.getValue().getUrl());
            PlatformTransactionManager transactionManager = new DataSourceTransactionManager(source.getValue());
            SpringTransactionPolicy transactionPolicy = new SpringTransactionPolicy(transactionManager);
            registry.registerBeanDefinition(source.getKey() + "TransactionPolicy", BeanDefinitionBuilder.genericBeanDefinition(SpringTransactionPolicy.class, () -> transactionPolicy).getBeanDefinition());
            registry.registerBeanDefinition(source.getKey() + "TransactionManager", BeanDefinitionBuilder.genericBeanDefinition(PlatformTransactionManager.class, () -> transactionManager).getBeanDefinition());
            registry.registerBeanDefinition(source.getKey(), BeanDefinitionBuilder.genericBeanDefinition(BasicDataSource.class, () -> source.getValue()).getBeanDefinition());
        }
    }

    @Override
    public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException { }

}

Я терял слух из-за этого последние пару дней и не могупройти мимо этого.Любая помощь или идеи будут с благодарностью.Я думаю, что это связано с транзакциями или с тем, как я создаю свои менеджеры транзакций, политики и источники данных.

Дополнительная информация:

• Spring Boot 2.1.1
• Верблюд 2.23.0
• Основным источником данных является Postgres 10.6

Answer 1

Я думаю, что решил это удовлетворительным образом.Это были транзакции, как я и подозревал.

Я добавил политику транзакций к своему целевому источнику данных:

@Override
public void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry registry) throws BeansException {
    registerTargetTransactionPolicy(datasources.get("target"), registry);
    for (Entry<String, BasicDataSource> source : datasources.entrySet()) {
        LOG.info("Adding datasource {} targeting {} to Spring registry", source.getKey(), source.getValue().getUrl());
        registry.registerBeanDefinition(source.getKey(), BeanDefinitionBuilder.genericBeanDefinition(BasicDataSource.class, () -> source.getValue()).getBeanDefinition());
    }
}

private void registerTargetTransactionPolicy(BasicDataSource datasource, BeanDefinitionRegistry registry) {
    PlatformTransactionManager transactionManager = new DataSourceTransactionManager(datasource);
    SpringTransactionPolicy transactionPolicy = new SpringTransactionPolicy(transactionManager);
    transactionPolicy.setPropagationBehaviorName("PROPAGATION_REQUIRES_NEW");
    registry.registerBeanDefinition(TARGET_REQUIRES_NEW_POLICY, BeanDefinitionBuilder.genericBeanDefinition(SpringTransactionPolicy.class, () -> transactionPolicy).getBeanDefinition());
}

Затем я использую ее в своем верблюжьем маршруте, используя .policy(ContextConfiguration.TARGET_REQUIRES_NEW_POLICY), и она работает безупречно.

Answer 2

Запрос на выбор не выполняется из-за ошибки:

отношение "article_flags" уже существует

Может быть, это поток поможет.Также я не думаю, что это связано с транзакциями.