This commit is contained in:
Clebert Suconic 2018-03-27 10:12:14 -04:00
commit 5d3ba98a11
3 changed files with 101 additions and 12 deletions

View File

@ -110,7 +110,7 @@ public abstract class AbstractJDBCDriver {
protected abstract void createSchema() throws SQLException;
protected final void createTable(String... schemaSqls) throws SQLException {
createTableIfNotExists(connection, sqlProvider.getTableName(), schemaSqls);
createTableIfNotExists(sqlProvider.getTableName(), schemaSqls);
}
private void connect() throws SQLException {
@ -175,9 +175,7 @@ public abstract class AbstractJDBCDriver {
}
}
private static void createTableIfNotExists(Connection connection,
String tableName,
String... sqls) throws SQLException {
private void createTableIfNotExists(String tableName, String... sqls) throws SQLException {
logger.tracef("Validating if table %s didn't exist before creating", tableName);
try {
connection.setAutoCommit(false);
@ -190,17 +188,27 @@ public abstract class AbstractJDBCDriver {
if (sqlWarning != null) {
logger.warn(JDBCUtils.appendSQLExceptionDetails(new StringBuilder(), sqlWarning));
}
try (Statement statement = connection.createStatement()) {
for (String sql : sqls) {
statement.executeUpdate(sql);
final SQLWarning statementSqlWarning = statement.getWarnings();
if (statementSqlWarning != null) {
logger.warn(JDBCUtils.appendSQLExceptionDetails(new StringBuilder(), statementSqlWarning, sql));
}
} else {
try (Statement statement = connection.createStatement();
ResultSet cntRs = statement.executeQuery(sqlProvider.getCountJournalRecordsSQL())) {
if (rs.next() && rs.getInt(1) > 0) {
logger.tracef("Table %s did exist but is not empty. Skipping initialization.", tableName);
} else {
sqls = Arrays.copyOfRange(sqls, 1, sqls.length);
}
}
}
try (Statement statement = connection.createStatement()) {
for (String sql : sqls) {
statement.executeUpdate(sql);
final SQLWarning statementSqlWarning = statement.getWarnings();
if (statementSqlWarning != null) {
logger.warn(JDBCUtils.appendSQLExceptionDetails(new StringBuilder(), statementSqlWarning, sql));
}
}
}
}
connection.commit();
} catch (SQLException e) {
final String sqlStatements = Stream.of(sqls).collect(Collectors.joining("\n"));

View File

@ -18,6 +18,8 @@
package org.apache.activemq.artemis.core.server.impl.jdbc;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@ -33,13 +35,31 @@ import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.junit.runners.Parameterized.Parameter;
@RunWith(Parameterized.class)
public class JdbcLeaseLockTest extends ActiveMQTestBase {
private JdbcSharedStateManager jdbcSharedStateManager;
private DatabaseStorageConfiguration dbConf;
private SQLProvider sqlProvider;
@Parameterized.Parameters(name = "create_tables_prior_test")
public static List<Object[]> data() {
return Arrays.asList(new Object[][] {
{true, null},
{false, null}
});
}
@Parameter(0)
public boolean withExistingTable;
@Parameter(1)
public Object result;
private LeaseLock lock() {
return lock(dbConf.getJdbcLockExpirationMillis());
}
@ -59,12 +79,23 @@ public class JdbcLeaseLockTest extends ActiveMQTestBase {
}
@Before
public void createLockTable() {
public void createLockTable() throws Exception {
dbConf = createDefaultDatabaseStorageConfiguration();
sqlProvider = JDBCUtils.getSQLProvider(
dbConf.getJdbcDriverClassName(),
dbConf.getNodeManagerStoreTableName(),
SQLProvider.DatabaseStoreType.NODE_MANAGER);
if (withExistingTable) {
TestJDBCDriver testDriver = TestJDBCDriver
.usingConnectionUrl(
dbConf.getJdbcConnectionUrl(),
dbConf.getJdbcDriverClassName(),
sqlProvider);
testDriver.start();
testDriver.stop();
}
jdbcSharedStateManager = JdbcSharedStateManager
.usingConnectionUrl(
UUID.randomUUID().toString(),

View File

@ -0,0 +1,50 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.core.server.impl.jdbc;
import java.sql.SQLException;
import org.apache.activemq.artemis.jdbc.store.drivers.AbstractJDBCDriver;
import org.apache.activemq.artemis.jdbc.store.sql.SQLProvider;
public class TestJDBCDriver extends AbstractJDBCDriver {
public static TestJDBCDriver usingConnectionUrl(
String jdbcConnectionUrl,
String jdbcDriverClass,
SQLProvider provider) {
TestJDBCDriver driver = new TestJDBCDriver();
driver.setSqlProvider(provider);
driver.setJdbcConnectionUrl(jdbcConnectionUrl);
driver.setJdbcDriverClass(jdbcDriverClass);
return driver;
}
@Override
protected void prepareStatements() throws SQLException {
}
@Override
protected void createSchema() throws SQLException {
try {
connection.createStatement().execute(sqlProvider.createNodeManagerStoreTableSQL());
} catch (SQLException e) {
}
}
}