Spring Курсор Batch Reader закрыт на ранней стадии управления транзакцией JTA

Рабочая конфигурация для рассматриваемого шага выглядит следующим образом:

Шаг, Spring Репозиторий пакетных заданий и бизнес-репозитории (с использованием различных источников данных) используют диспетчер транзакций JTA.
Шаг "myStep" использует считыватель чейков Jdbc.
WebLogic, Oracle XE и/или EE

Мне хотелось проанализировать производительность JDBC Cursor Item Reader в "myStep" , однако после первого коммита второй прочитанный первый фрагмент завершится с ошибкой с помощью java.sql.SQLException: результат уже установлен замкнутый.

Я подозревал, что по какой-то причине драйвер JTA/XA закрывает курсор, поэтому я дал "myStep" простой диспетчер транзакций данных (на источнике данных, который использовал читатель), и этот шаг был успешно завершен. Это не решение, так как это прерывает транзакционную целостность шага.

Должен ли я использовать читатель курсора внутри управляемого шага JTA (используя среду, описанную ниже)? Если да, то что может быть неправильно настроено на моем конце?

Окружающая среда

Менеджер транзакций: <bean id="myTransactionManager" class="org.springframework.transaction.jta.JtaTransactionManager"/>
Драйвер источника данных: OracleXADataSource JDBC 6 11.1.0.7.0
WebLogic: 12.1.3.0.0
Oracle DB 11g: Enterprise Edition 11.2.0.4.0
ОС: OSX или Linux

Config

<bean id="myTransactionManager" class="org.springframework.transaction.jta.JtaTransactionManager"/>

<bean id="myDataSource" class="org.springframework.jndi.JndiObjectFactoryBean">
    <property name="jndiName" value="jdbc/myDataSource"/>
    <property name="proxyInterface" value="javax.sql.DataSource"/>
</bean>

<batch:step id="myStep" job-repository="myJobRepositoryFactory">
    <batch:tasklet transaction-manager="myTransactionManager">
        <batch:chunk
                reader="myReader"
                processor="myProcessor"
                writer="myWriter"
                commit-interval="100"
                processor-transactional="false"/>
        <batch:listeners>
            <batch:listener ref="myListener"/>
        </batch:listeners>
    </batch:tasklet>
</batch:step>

<bean id="myReader" class="org.springframework.batch.item.database.JdbcCursorItemReader" scope="step">
    <property name="dataSource" ref="myDataSource"/>
    <property name="sql" value="SELECT * FROM myHugeTable ORDER BY myColumn DESC"/>
    <property name="rowMapper">
        <bean class="myRowMapper"/>
    </property>
</bean>

Пойманный в действии

Ниже приведен стек вызовов из результирующего набора, который закрывается до следующего чтения chunk. Обратите внимание, что XA Connection закрывает все утверждения, что заставляет JDBC закрывать все наборы результатов.

java.lang.Thread.State: RUNNABLE
  at weblogic.jdbc.wrapper.ResultSet.internalClose(ResultSet.java:178)
  at weblogic.jdbc.wrapper.Statement.closeAllResultSets(Statement.java:286)
  at weblogic.jdbc.wrapper.Statement.internalClose(Statement.java:395)
  at weblogic.jdbc.wrapper.Statement.internalClose(Statement.java:367)
  at weblogic.jdbc.wrapper.XAConnection.closeAllStatements(XAConnection.java:393)
  at weblogic.jdbc.wrapper.XAConnection.cleanup(XAConnection.java:406)
  at weblogic.jdbc.wrapper.XAConnection.releaseToPool(XAConnection.java:432)
  at weblogic.jdbc.jta.DataSource.removeTxAssoc(DataSource.java:1907)
  at weblogic.jdbc.jta.DataSource.prepare(DataSource.java:1090)
  at weblogic.transaction.internal.XAServerResourceInfo.prepare(XAServerResourceInfo.java:1408)
  at weblogic.transaction.internal.XAServerResourceInfo.prepare(XAServerResourceInfo.java:522)
  at weblogic.transaction.internal.ServerSCInfo.startPrepare(ServerSCInfo.java:411)
  at weblogic.transaction.internal.ServerTransactionImpl.localPrepare(ServerTransactionImpl.java:2709)
  at weblogic.transaction.internal.ServerTransactionImpl.globalPrepare(ServerTransactionImpl.java:2340)
  at weblogic.transaction.internal.ServerTransactionImpl.internalCommit(ServerTransactionImpl.java:300)
  at weblogic.transaction.internal.ServerTransactionImpl.commit(ServerTransactionImpl.java:260)
  at org.glassfish.transaction.TransactionManagerImplCommon.commit(TransactionManagerImplCommon.java:571)
  at org.springframework.transaction.jta.JtaTransactionManager.doCommit(JtaTransactionManager.java:1021)
  at org.springframework.transaction.support.AbstractPlatformTransactionManager.processCommit(AbstractPlatformTransactionManager.java:761)
  at org.springframework.transaction.support.AbstractPlatformTransactionManager.commit(AbstractPlatformTransactionManager.java:730)
  at org.springframework.transaction.support.TransactionTemplate.execute(TransactionTemplate.java:150)
  at org.springframework.batch.core.step.tasklet.TaskletStep$2.doInChunkContext(TaskletStep.java:271)
  at org.springframework.batch.core.scope.context.StepContextRepeatCallback.doInIteration(StepContextRepeatCallback.java:77)
  at org.springframework.batch.repeat.support.RepeatTemplate.getNextResult(RepeatTemplate.java:368)
  at org.springframework.batch.repeat.support.RepeatTemplate.executeInternal(RepeatTemplate.java:215)
  at org.springframework.batch.repeat.support.RepeatTemplate.iterate(RepeatTemplate.java:144)
  at org.springframework.batch.core.step.tasklet.TaskletStep.doExecute(TaskletStep.java:257)
  at org.springframework.batch.core.step.AbstractStep.execute(AbstractStep.java:198)
  at org.springframework.batch.core.job.SimpleStepHandler.handleStep(SimpleStepHandler.java:148)
  at org.springframework.batch.core.job.flow.JobFlowExecutor.executeStep(JobFlowExecutor.java:64)
  at org.springframework.batch.core.job.flow.support.state.StepState.handle(StepState.java:67)
  at org.springframework.batch.core.job.flow.support.SimpleFlow.resume(SimpleFlow.java:165)
  at org.springframework.batch.core.job.flow.support.SimpleFlow.start(SimpleFlow.java:144)
  at org.springframework.batch.core.job.flow.FlowJob.doExecute(FlowJob.java:134)
  at org.springframework.batch.core.job.AbstractJob.execute(AbstractJob.java:304)
  at org.springframework.batch.core.launch.support.SimpleJobLauncher$1.run(SimpleJobLauncher.java:135)
  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
  at java.lang.Thread.run(Thread.java:745)

Ответ 1

Вы должны иметь возможность использовать курсор-читатель внутри управляемого шага JTA. Мы делаем именно это в проекте, над которым я работаю. Мы используем Atomikos как XA TM.

Вот наша конфигурация XA/JTA, которую мы используем. Возможно, это для вас полезно:

@Bean(initMethod = "init", destroyMethod = "shutdownForce")
public UserTransactionService userTransactionService() {
    return new UserTransactionServiceImp(userTransactionServiceProperties());
}

@Bean(initMethod = "init", destroyMethod = "close")
@DependsOn("userTransactionService")
public UserTransactionManager atomikosTransactionManager() {
    UserTransactionManager userTransactionManager = new UserTransactionManager();
    userTransactionManager.setForceShutdown(true);
    userTransactionManager.setStartupTransactionService(false);
    return userTransactionManager;
}

@Bean
@DependsOn("userTransactionService")
public UserTransaction atomikosUserTransaction() throws SystemException {
    return new UserTransactionImp();
}

@Bean
@DependsOn("userTransactionService")
public JtaTransactionManager transactionManager() throws SystemException {
    JtaTransactionManager jtaTransactionManager = new JtaTransactionManager();
    jtaTransactionManager.setTransactionManager(atomikosTransactionManager());
    jtaTransactionManager.setUserTransaction(atomikosUserTransaction());
    jtaTransactionManager.setAllowCustomIsolationLevels(true);
    return jtaTransactionManager;
}

Все наши источники данных создаются как org.springframework.boot.jta.atomikos.AtomikosDataSourceBean. Например, источник Ora-данных создается таким образом:

    AtomikosDataSourceBean oraXaDs = new AtomikosDataSourceBean();
    oraXaDs.setXaDataSourceClassName(oraDsProp.getDatasourceClass());
    oraXaDs.setUniqueResourceName(oraDsProp.getInstancename());
    oraXaDs.setMinPoolSize(oraDsProp.getPoolMinSize());
    oraXaDs.setMaxPoolSize(oraDsProp.getPoolMaxSize());
    oraXaDs.setTestQuery(oraDsProp.getValidConnectionSQL());

    Properties oraXaDsProps = oraXaDs.getXaProperties();
    oraXaDsProps.setProperty("user", oraDsProp.getUser());
    oraXaDsProps.setProperty("password", oraDsProp.getPassword());
    oraXaDsProps.setProperty("URL", oraDsProp.getUrl());

Ответ 2

Мои два цента по этой проблеме:

Сначала прозрение:

Чтение из курсора базы данных означает открытие соединения, запуск одного оператора SQL против него и постоянное чтение строк во время всего пакетного задания. Это имеет смысл, потому что часто вводные данные задания могут быть охарактеризованы одним оператором SQL, но выполнение его и чтение всех данных с аванса ResultSet, конечно, не является решением. У нас просто есть одна проблема с постоянным чтением: фиксация транзакции закроет соединение. Итак, как мы держим его открытым? Простое решение: оно не участвует в транзакции. Spring Пакеты JdbcCursorItemReader использует отдельное соединение для открытия курсора, тем самым минуя транзакцию, управляемую диспетчером транзакций. В среде сервера приложений мы должны сделать немного больше, чтобы заставить его работать. Обычно мы получаем соединения с DataSource, управляемыми сервером приложений, и все эти соединения принимают участие в транзакциях по умолчанию. Нам нужно настроить отдельный DataSource, который не участвует в транзакциях, и вводить его только в наши курсоры. Включение их в другое место может нанести большой ущерб безопасности транзакций.

Ваша проблема в вашем шаге в основном: (из того, что я могу заключить, не просматривая файл datasource xml:))

Шаг, Spring Репозиторий пакетных заданий и бизнес-репозитории (с использованием различных источников данных) используют диспетчер транзакций JTA.

Менеджер транзакций JTA Spring должен использоваться таким образом, чтобы weblogic обрабатывал ваши транзакции JTA. Вместо этого вам нужно настроить только источник данных (который также должен находиться в контейнере) для использования драйверов XA и в вашем приложении, используйте соответствующий диспетчер транзакций и найдите источники данных, поддерживающие XA, JNDI

Однако, если вы не можете полагаться на контейнер (например, вы пишете автономное приложение и т.д.), вам нужно будет найти управляющий транзакциями, который способен на это. Atomikos - одна из самых известных бесплатных библиотек JTA/XA.

Сказав, что после того, как вы настроили либо путь JNDI, либо способ Atomikos, вот конфигурация, которую следует иметь в виду при использовании нескольких источников данных:

<batch:tasklet>
  <batch:transaction-attributes isolation="READ_COMMITTED" propagation="REQUIRES_NEW" timeout="200"/>
  <batch:chunk reader="myItemReader" writer="myItemWriter" commit-interval="20"/>
</batch:tasklet>

Надеюсь, что это очистит воздух от этой проблемы.