.classpath | 1
modules/enterprise/server/data-migration/pom.xml | 14
modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/DataMigratorRunner.java | 72 +--
modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/ExistingDataBulkExportSource.java | 124 -----
modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/ExistingDataJPABulkExportSource.java | 58 --
modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/ExistingDataJPASource.java | 64 --
modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/ExistingDataSource.java | 36 -
modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/ExistingPostgresDataBulkExportSource.java | 80 ---
modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/ScrollableDataSource.java | 156 ------
modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/Telemetry.java | 57 --
modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/datasources/ExistingDataBulkExportSource.java | 124 +++++
modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/datasources/ExistingDataJPABulkExportSource.java | 58 ++
modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/datasources/ExistingDataJPASource.java | 64 ++
modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/datasources/ExistingDataSource.java | 36 +
modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/datasources/ExistingPostgresDataBulkExportSource.java | 80 +++
modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/datasources/ScrollableDataSource.java | 157 ++++++
modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/workers/AbstractMigrationWorker.java | 6
modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/workers/AggregateDataMigrator.java | 3
modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/workers/RawDataMigrator.java | 5
modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/workers/Telemetry.java | 57 ++
modules/enterprise/server/data-migration/src/test/java/org/rhq/server/metrics/DataSourceTest.java | 169 -------
modules/enterprise/server/data-migration/src/test/java/org/rhq/server/metrics/migrator/DataMigratorTest.java | 235 ++++++++++
modules/enterprise/server/data-migration/src/test/java/org/rhq/server/metrics/migrator/DataSourceTest.java | 169 +++++++
modules/enterprise/server/data-migration/src/test/java/org/rhq/server/metrics/migrator/workers/AggregateDataMigratorTest.java | 198 ++++++++
modules/enterprise/server/data-migration/src/test/java/org/rhq/server/metrics/migrator/workers/RawDataMigratorTest.java | 194 ++++++++
25 files changed, 1436 insertions(+), 781 deletions(-)
New commits:
commit d9014d8f146a01a76a5581bf56fc97d7f9c79e21
Author: Stefan Negrea <snegrea(a)redhat.com>
Date: Wed Jun 26 13:20:42 2013 -0500
[BZ 961361] The estimation process was wrongly attempting to estimate actual data migratio when only deletion was requested by the user.
diff --git a/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/DataMigratorRunner.java b/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/DataMigratorRunner.java
index b2fe3d3..7eef2a3 100644
--- a/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/DataMigratorRunner.java
+++ b/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/DataMigratorRunner.java
@@ -506,10 +506,10 @@ public class DataMigratorRunner {
}
} else {
migrator.deleteAllDataAtEndOfMigration();
- migrator.runRawDataMigration(true);
- migrator.run1HAggregateDataMigration(true);
- migrator.run6HAggregateDataMigration(true);
- migrator.run1DAggregateDataMigration(true);
+ migrator.runRawDataMigration(false);
+ migrator.run1HAggregateDataMigration(false);
+ migrator.run6HAggregateDataMigration(false);
+ migrator.run1DAggregateDataMigration(false);
System.out.println("Estimation process - starting\n");
long estimate = migrator.estimate();
@@ -518,6 +518,11 @@ public class DataMigratorRunner {
System.out.println("Estimation process - ended\n\n");
if (!(Boolean) configuration.get(estimateOnlyOption)) {
+ migrator.runRawDataMigration(true);
+ migrator.run1HAggregateDataMigration(true);
+ migrator.run6HAggregateDataMigration(true);
+ migrator.run1DAggregateDataMigration(true);
+
System.out.println("Old data deletion process - starting\n");
long startTime = System.currentTimeMillis();
migrator.deleteOldData();
commit 3dcdd4e454ae76bd75854d78e2d2bbffefa50475
Author: Stefan Negrea <snegrea(a)redhat.com>
Date: Wed Jun 26 13:10:05 2013 -0500
Updates to the data migrator:
1) remove the preserve data option since it can be confusing to end users
2) update the way data preservation is trigger via the delete data option
3) add code to close both the SQL and Cassandra sessions prior to program completion.
diff --git a/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/DataMigratorRunner.java b/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/DataMigratorRunner.java
index 86cb2fd..b2fe3d3 100644
--- a/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/DataMigratorRunner.java
+++ b/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/DataMigratorRunner.java
@@ -23,8 +23,6 @@ package org.rhq.server.metrics.migrator;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
-import java.sql.Driver;
-import java.util.Enumeration;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
@@ -118,8 +116,6 @@ public class DataMigratorRunner {
.withDescription("Disable 6 hours aggregates table migration (default: false)").create();
private Option disable1DOption = OptionBuilder.withLongOpt("disable-1d-migration").hasOptionalArg().withType(Boolean.class)
.withDescription("Disable 24 hours aggregates table migration (default: false)").create();
- private Option preserveDataOption = OptionBuilder.withLongOpt("preserve-data").hasOptionalArg().withType(Boolean.class)
- .withDescription("Preserve SQL data post migration (default: true)").create();
private Option deleteDataOption = OptionBuilder.withLongOpt("delete-data").hasOptionalArg().withType(Boolean.class)
.withDescription("Delete SQL data at the end of migration (default: false)").create();
private Option estimateOnlyOption = OptionBuilder.withLongOpt("estimate-only").hasOptionalArg().withType(Boolean.class)
@@ -215,7 +211,6 @@ public class DataMigratorRunner {
options.addOption(disable1HOption);
options.addOption(disable6HOption);
options.addOption(disable1DOption);
- options.addOption(preserveDataOption);
options.addOption(deleteDataOption);
options.addOption(estimateOnlyOption);
options.addOption(deleteOnlyOption);
@@ -280,8 +275,8 @@ public class DataMigratorRunner {
configuration.put(disable1HOption, false);
configuration.put(disable6HOption, false);
configuration.put(disable1DOption, false);
- configuration.put(preserveDataOption, true);
configuration.put(estimateOnlyOption, false);
+ configuration.put(deleteDataOption, false);
configuration.put(deleteOnlyOption, false);
configuration.put(experimentalExportOption, false);
}
@@ -449,12 +444,9 @@ public class DataMigratorRunner {
configuration.put(disable1DOption, value);
}
- if (commandLine.hasOption(preserveDataOption.getLongOpt())) {
- value = tryParseBoolean(commandLine.getOptionValue(preserveDataOption.getLongOpt()), true);
- configuration.put(preserveDataOption, value);
- } else if (commandLine.hasOption(deleteDataOption.getLongOpt())) {
+ if (commandLine.hasOption(deleteDataOption.getLongOpt())) {
value = tryParseBoolean(commandLine.getOptionValue(deleteDataOption.getLongOpt()), true);
- configuration.put(preserveDataOption, value);
+ configuration.put(deleteDataOption, value);
}
if (commandLine.hasOption(estimateOnlyOption.getLongOpt())) {
@@ -470,25 +462,26 @@ public class DataMigratorRunner {
private void run() throws Exception {
log.debug("Creating Entity Manager");
- EntityManager entityManager = this.createEntityManager();
+ EntityManagerFactory entityManagerFactory = this.createEntityManagerFactory();
+ EntityManager entityManager = entityManagerFactory.createEntityManager();
log.debug("Done creating Entity Manager");
log.debug("Creating Cassandra session");
- Session session = this.createCassandraSession();
+ Session cassandraSession = this.createCassandraSession();
log.debug("Done creating Cassandra session");
DatabaseType databaseType = DatabaseType.Postgres;
if ("oracle".equals(configuration.get(sqlServerType))) {
databaseType = databaseType.Oracle;
}
- DataMigrator migrator = new DataMigrator(entityManager, session, databaseType, tryParseBoolean(
+ DataMigrator migrator = new DataMigrator(entityManager, cassandraSession, databaseType, tryParseBoolean(
configuration.get(experimentalExportOption), false));
if (!(Boolean) configuration.get(deleteOnlyOption)) {
- if ((Boolean) configuration.get(preserveDataOption)) {
- migrator.preserveData();
- } else {
+ if ((Boolean) configuration.get(deleteDataOption)) {
migrator.deleteAllDataAtEndOfMigration();
+ } else {
+ migrator.preserveData();
}
migrator.runRawDataMigration(!(Boolean) configuration.get(disableRawOption));
@@ -535,12 +528,28 @@ public class DataMigratorRunner {
}
}
- /*if (entityManager != null) {
- entityManager.close();
- }
+ this.closeCassandraSession(cassandraSession);
+ this.closeEntityManagerFactory(entityManagerFactory);
+ }
+
+ /**
+ * Closes an existing cassandra session.
+ *
+ * @param session
+ */
+ private void closeCassandraSession(Session session) {
+ session.shutdown();
+ }
+
+ /**
+ * Closes an existing entity manager factory
+ *
+ * @param entityManagerFactory
+ */
+ private void closeEntityManagerFactory(EntityManagerFactory entityManagerFactory) {
if (entityManagerFactory != null) {
entityManagerFactory.close();
- }*/
+ }
}
/**
@@ -569,12 +578,12 @@ public class DataMigratorRunner {
}
/**
- * Create a hibernate session to the SQL server.
+ * Create an entity manager factory with the SQL configuration from properties.
*
- * @return
+ * @return an entity manager factory
* @throws Exception
*/
- private EntityManager createEntityManager() throws Exception {
+ private EntityManagerFactory createEntityManagerFactory() throws Exception {
Properties properties = new Properties();
properties.put("javax.persistence.provider", "org.hibernate.ejb.HibernatePersistence");
properties.put("hibernate.connection.username", (String) configuration.get(sqlUserOption));
@@ -624,7 +633,7 @@ public class DataMigratorRunner {
Ejb3Configuration configuration = new Ejb3Configuration();
configuration.setProperties(properties);
EntityManagerFactory factory = configuration.buildEntityManagerFactory();
- return factory.createEntityManager();
+ return factory;
}
/**
commit cff10e45e97c4517cd90e9cbca0c35c8e4f1e0b2
Author: Stefan Negrea <snegrea(a)redhat.com>
Date: Wed Jun 26 12:32:36 2013 -0500
Add unit tests to the data migrator. Also, reorganized the structure of the data migration package.
The set of unit tests incorporates an initial set created by: Armine Hovsepyan <ahovsepy(a)redhat.com>
diff --git a/.classpath b/.classpath
index d46b67d..70d5f0a 100644
--- a/.classpath
+++ b/.classpath
@@ -215,6 +215,7 @@
<classpathentry kind="src" path="modules/common/cassandra-jmx/src/main/java"/>
<classpathentry kind="src" path="modules/plugins/rhq-storage/src/main/java"/>
<classpathentry kind="src" path="modules/enterprise/server/data-migration/src/main/java"/>
+ <classpathentry kind="src" path="modules/enterprise/server/data-migration/src/test/java"/>
<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER/org.eclipse.jdt.internal.debug.ui.launcher.StandardVMType/JavaSE-1.7"/>
<classpathentry exported="true" kind="var" path="M2_REPO/org/apache/httpcomponents/httpclient/4.2.3/httpclient-4.2.3.jar" sourcepath="M2_REPO/org/apache/httpcomponents/httpclient/4.2.3/httpclient-4.2.3-sources.jar"/>
<classpathentry exported="true" kind="var" path="M2_REPO/org/apache/httpcomponents/httpcore/4.2.2/httpcore-4.2.2.jar" sourcepath="M2_REPO/org/apache/httpcomponents/httpcore/4.2.2/httpcore-4.2.2-sources.jar"/>
diff --git a/modules/enterprise/server/data-migration/pom.xml b/modules/enterprise/server/data-migration/pom.xml
index 83aedc5..ea25952 100644
--- a/modules/enterprise/server/data-migration/pom.xml
+++ b/modules/enterprise/server/data-migration/pom.xml
@@ -84,6 +84,20 @@
<artifactId>joda-time</artifactId>
<version>${joda-time.version}</version>
</dependency>
+
+ <dependency>
+ <groupId>org.powermock</groupId>
+ <artifactId>powermock-module-testng</artifactId>
+ <version>${powermock.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.powermock</groupId>
+ <artifactId>powermock-api-mockito</artifactId>
+ <version>${powermock.version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
diff --git a/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/ExistingDataBulkExportSource.java b/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/ExistingDataBulkExportSource.java
deleted file mode 100644
index 756d1b1..0000000
--- a/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/ExistingDataBulkExportSource.java
+++ /dev/null
@@ -1,124 +0,0 @@
-/*
- * RHQ Management Platform
- * Copyright (C) 2005-2013 Red Hat, Inc.
- * All rights reserved.
- *
- * This program is free software; you can redistribute it and/or modify
- * it under the terms of the GNU General Public License as published by
- * the Free Software Foundation version 2 of the License.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
- * GNU General Public License for more details.
- *
- * You should have received a copy of the GNU General Public License
- * along with this program; if not, write to the Free Software Foundation, Inc.,
- * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA
- */
-
-package org.rhq.server.metrics.migrator;
-
-import java.io.BufferedReader;
-import java.io.File;
-import java.io.FileReader;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Random;
-import java.util.StringTokenizer;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-
-import org.rhq.core.util.stream.StreamUtil;
-
-/**
- * @author Thomas Segismont
- */
-public abstract class ExistingDataBulkExportSource implements ExistingDataSource {
-
- private static final Log log = LogFactory.getLog(ExistingDataBulkExportSource.class);
-
- protected static final int IO_BUFFER_SIZE = 1024 * 64;
-
- protected static final String DELIMITER = "|";
-
- private File existingDataFile;
-
- private BufferedReader existingDataFileReader;
-
- private int currentIndex;
-
- public ExistingDataBulkExportSource() {
- Random random = new Random();
- String exportFileName = System.currentTimeMillis() + "." + random.nextInt() + ".export";
- this.existingDataFile = new File(System.getProperty("java.io.tmpdir"), exportFileName);
- }
-
- @Override
- public void initialize() throws Exception {
- this.exportExistingData();
- this.startReading();
- }
-
- @Override
- public void close() throws Exception {
- this.stopReading();
- this.removeTempFile();
- }
-
- protected abstract void exportExistingData() throws Exception;
-
- protected File getExistingDataFile() {
- return existingDataFile;
- }
-
- protected void startReading() throws Exception {
- if (!existingDataFile.exists() && !existingDataFile.isFile() && !existingDataFile.canRead()) {
- throw new IllegalStateException();
- }
- existingDataFileReader = new BufferedReader(new FileReader(existingDataFile));
- currentIndex = 0;
- }
-
- protected void stopReading() {
- StreamUtil.safeClose(existingDataFileReader);
- }
-
- protected void removeTempFile(){
- if(this.existingDataFile.exists()){
- try{
- this.existingDataFile.delete();
- }catch(Exception e){
- log.debug("Unable to clean temporary file " + this.existingDataFile, e);
- }
- }
- }
-
- @Override
- public List<Object[]> getData(int fromIndex, int maxResults) throws Exception {
- if (log.isDebugEnabled()) {
- log.debug("Reading lines " + fromIndex + " to " + (fromIndex + maxResults));
- }
-
- if (fromIndex != currentIndex) {
- throw new IllegalStateException();
- }
-
- List<Object[]> results = new LinkedList<Object[]>();
- for (int i = 0; i < maxResults; i++) {
- String nextLine = existingDataFileReader.readLine();
- if (nextLine == null) {
- break;
- }
- currentIndex++;
- StringTokenizer stringTokenizer = new StringTokenizer(nextLine, DELIMITER);
- Object[] row = new Object[stringTokenizer.countTokens()];
- for (int j = 0; j < row.length; j++) {
- row[j] = stringTokenizer.nextToken();
- }
- results.add(row);
- }
- return results;
- }
-}
diff --git a/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/ExistingDataJPABulkExportSource.java b/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/ExistingDataJPABulkExportSource.java
deleted file mode 100644
index fde9096..0000000
--- a/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/ExistingDataJPABulkExportSource.java
+++ /dev/null
@@ -1,58 +0,0 @@
-package org.rhq.server.metrics.migrator;
-
-import java.io.BufferedWriter;
-import java.io.FileWriter;
-import java.sql.Connection;
-import java.sql.PreparedStatement;
-import java.sql.ResultSet;
-
-import javax.persistence.EntityManager;
-
-import org.hibernate.Session;
-import org.hibernate.engine.spi.SessionFactoryImplementor;
-import org.hibernate.service.jdbc.connections.spi.ConnectionProvider;
-
-import org.rhq.core.util.jdbc.JDBCUtil;
-import org.rhq.core.util.stream.StreamUtil;
-
-/**
- * @author Thomas Segismont
- */
-public class ExistingDataJPABulkExportSource extends ExistingDataBulkExportSource {
-
- private EntityManager entityManager;
- private String selectNativeQuery;
-
- public ExistingDataJPABulkExportSource(EntityManager entityManager, String selectNativeQuery) {
- super();
- }
-
- protected void exportExistingData() throws Exception {
- BufferedWriter fileWriter = null;
- Connection connection = null;
- PreparedStatement statement = null;
- ResultSet resultSet = null;
- try {
- fileWriter = new BufferedWriter(new FileWriter(getExistingDataFile()), IO_BUFFER_SIZE);
- Session session = (Session) this.entityManager.getDelegate();
- SessionFactoryImplementor sfi = (SessionFactoryImplementor) session.getSessionFactory();
- ConnectionProvider cp = sfi.getConnectionProvider();
- connection = cp.getConnection();
- statement = connection.prepareStatement(this.selectNativeQuery);
- resultSet = statement.executeQuery();
- int columnCount = resultSet.getMetaData().getColumnCount();
- while (resultSet.next()) {
- for (int i = 1; i < columnCount + 1; i++) {
- if (i > 1) {
- fileWriter.write(DELIMITER);
- }
- fileWriter.write(resultSet.getString(i));
- }
- fileWriter.write("\n");
- }
- } finally {
- StreamUtil.safeClose(fileWriter);
- JDBCUtil.safeClose(connection, statement, resultSet);
- }
- }
-}
diff --git a/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/ExistingDataJPASource.java b/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/ExistingDataJPASource.java
deleted file mode 100644
index 9aca36a..0000000
--- a/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/ExistingDataJPASource.java
+++ /dev/null
@@ -1,64 +0,0 @@
-/*
- * RHQ Management Platform
- * Copyright (C) 2005-2013 Red Hat, Inc.
- * All rights reserved.
- *
- * This program is free software; you can redistribute it and/or modify
- * it under the terms of the GNU General Public License as published by
- * the Free Software Foundation version 2 of the License.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
- * GNU General Public License for more details.
- *
- * You should have received a copy of the GNU General Public License
- * along with this program; if not, write to the Free Software Foundation, Inc.,
- * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA
- */
-
-package org.rhq.server.metrics.migrator;
-
-import java.util.List;
-
-import javax.persistence.EntityManager;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-
-/**
- * @author Thomas Segismont
- */
-public class ExistingDataJPASource implements ExistingDataSource {
-
- private static final Log log = LogFactory.getLog(ExistingDataJPASource.class);
-
- private EntityManager entityManager;
- private String selectNativeQuery;
-
- public ExistingDataJPASource(EntityManager entityManager, String selectNativeQuery) {
- this.entityManager = entityManager;
- this.selectNativeQuery = selectNativeQuery;
- }
-
- @SuppressWarnings("unchecked")
- @Override
- public List<Object[]> getData(int fromIndex, int maxResults) throws Exception {
- if (log.isDebugEnabled()) {
- log.debug("Reading lines " + fromIndex + " to " + (fromIndex + maxResults));
- }
-
- return (List<Object[]>) entityManager.createNativeQuery(selectNativeQuery).setFirstResult(fromIndex)
- .setMaxResults(maxResults).getResultList();
- }
-
- @Override
- public void initialize() {
- //nothing to do since it just implements a simple query with limits
- }
-
- @Override
- public void close() {
- //nothing to do since it just implements a simple query with limits
- }
-}
diff --git a/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/ExistingDataSource.java b/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/ExistingDataSource.java
deleted file mode 100644
index 033c5d9..0000000
--- a/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/ExistingDataSource.java
+++ /dev/null
@@ -1,36 +0,0 @@
-/*
- * RHQ Management Platform
- * Copyright (C) 2005-2013 Red Hat, Inc.
- * All rights reserved.
- *
- * This program is free software; you can redistribute it and/or modify
- * it under the terms of the GNU General Public License as published by
- * the Free Software Foundation version 2 of the License.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
- * GNU General Public License for more details.
- *
- * You should have received a copy of the GNU General Public License
- * along with this program; if not, write to the Free Software Foundation, Inc.,
- * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA
- */
-
-package org.rhq.server.metrics.migrator;
-
-import java.util.List;
-
-/**
- * @author Thomas Segismont
- */
-public interface ExistingDataSource {
-
- static final int TIMEOUT = 6000000;
-
- List<Object[]> getData(int fromIndex, int maxResults) throws Exception;
-
- void initialize() throws Exception;
-
- void close() throws Exception;
-}
diff --git a/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/ExistingPostgresDataBulkExportSource.java b/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/ExistingPostgresDataBulkExportSource.java
deleted file mode 100644
index 2ba7e4a..0000000
--- a/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/ExistingPostgresDataBulkExportSource.java
+++ /dev/null
@@ -1,80 +0,0 @@
-/*
- * RHQ Management Platform
- * Copyright (C) 2005-2013 Red Hat, Inc.
- * All rights reserved.
- *
- * This program is free software; you can redistribute it and/or modify
- * it under the terms of the GNU General Public License as published by
- * the Free Software Foundation version 2 of the License.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
- * GNU General Public License for more details.
- *
- * You should have received a copy of the GNU General Public License
- * along with this program; if not, write to the Free Software Foundation, Inc.,
- * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA
- */
-
-package org.rhq.server.metrics.migrator;
-
-import java.io.BufferedWriter;
-import java.io.FileWriter;
-import java.sql.Connection;
-
-import javax.persistence.EntityManager;
-
-import org.hibernate.Session;
-import org.hibernate.engine.spi.SessionFactoryImplementor;
-import org.hibernate.service.jdbc.connections.spi.ConnectionProvider;
-import org.postgresql.copy.CopyManager;
-import org.postgresql.core.BaseConnection;
-
-import org.rhq.core.util.jdbc.JDBCUtil;
-import org.rhq.core.util.stream.StreamUtil;
-
-/**
- * @author Thomas Segismont
- */
-public class ExistingPostgresDataBulkExportSource extends ExistingDataBulkExportSource {
-
- private final EntityManager entityManager;
- private final String selectNativeQuery;
- private final int maxResults;
-
- public ExistingPostgresDataBulkExportSource(EntityManager entityManager, String selectNativeQuery) {
- this(entityManager, selectNativeQuery, -1);
- }
-
- public ExistingPostgresDataBulkExportSource(EntityManager entityManager, String selectNativeQuery, int maxResults) {
- super();
- this.entityManager = entityManager;
- this.selectNativeQuery = selectNativeQuery;
- this.maxResults = maxResults;
- }
-
- protected void exportExistingData() throws Exception {
- BufferedWriter fileWriter = null;
- Connection connection = null;
- try {
- fileWriter = new BufferedWriter(new FileWriter(getExistingDataFile()), IO_BUFFER_SIZE);
- Session session = (Session) entityManager.getDelegate();
- SessionFactoryImplementor sfi = (SessionFactoryImplementor) session.getSessionFactory();
- ConnectionProvider cp = sfi.getConnectionProvider();
- connection = cp.getConnection();
- CopyManager copyManager = new CopyManager((BaseConnection) connection);
-
- if (maxResults > 0) {
- copyManager.copyOut("COPY (" + selectNativeQuery + " LIMIT " + maxResults
- + ") TO STDOUT WITH DELIMITER '" + DELIMITER + "'", fileWriter);
- } else {
- copyManager.copyOut("COPY (" + selectNativeQuery + ") TO STDOUT WITH DELIMITER '" + DELIMITER + "'",
- fileWriter);
- }
- } finally {
- StreamUtil.safeClose(fileWriter);
- JDBCUtil.safeClose(connection);
- }
- }
-}
diff --git a/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/ScrollableDataSource.java b/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/ScrollableDataSource.java
deleted file mode 100644
index 42161a4..0000000
--- a/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/ScrollableDataSource.java
+++ /dev/null
@@ -1,156 +0,0 @@
-/*
- * RHQ Management Platform
- * Copyright (C) 2005-2013 Red Hat, Inc.
- * All rights reserved.
- *
- * This program is free software; you can redistribute it and/or modify
- * it under the terms of the GNU General Public License as published by
- * the Free Software Foundation version 2 of the License.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
- * GNU General Public License for more details.
- *
- * You should have received a copy of the GNU General Public License
- * along with this program; if not, write to the Free Software Foundation, Inc.,
- * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA
- */
-
-package org.rhq.server.metrics.migrator;
-
-import java.util.ArrayList;
-import java.util.List;
-
-import javax.persistence.EntityManager;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.hibernate.Query;
-import org.hibernate.ScrollMode;
-import org.hibernate.ScrollableResults;
-import org.hibernate.Session;
-import org.hibernate.StatelessSession;
-
-import org.rhq.server.metrics.migrator.DataMigrator.DatabaseType;
-
-/**
- * @author Stefan Negrea
- */
-public class ScrollableDataSource implements ExistingDataSource {
-
- private static final Log log = LogFactory.getLog(ScrollableDataSource.class);
-
- private final EntityManager entityManager;
- private final DatabaseType databaseType;
- private final String selectNativeQuery;
- private final int maxResults;
-
- private ScrollableResults results;
- private StatelessSession session;
- private int lastMigratedItemIndex;
-
-
- public ScrollableDataSource(EntityManager entityManager, DatabaseType databaseType,
- String selectNativeQuery) {
- this(entityManager, databaseType, selectNativeQuery, -1);
- }
-
- public ScrollableDataSource(EntityManager entityManager, DatabaseType databaseType, String selectNativeQuery,
- int maxResults) {
- this.entityManager = entityManager;
- this.databaseType = databaseType;
- this.selectNativeQuery = selectNativeQuery;
- this.maxResults = maxResults;
- }
-
- @Override
- public List<Object[]> getData(int fromIndex, int maxResults) throws Exception {
- if (fromIndex != lastMigratedItemIndex + 1) {
- throw new Exception("Cursor error. " + fromIndex + " " + lastMigratedItemIndex);
- }
-
- if (log.isDebugEnabled()) {
- log.debug("Reading lines " + fromIndex + " to " + (fromIndex + maxResults));
- }
-
- List<Object[]> resultList = new ArrayList<Object[]>();
-
- int maxLimit = lastMigratedItemIndex + maxResults;
-
- while (results.next()) {
- resultList.add(results.get());
-
- if (++lastMigratedItemIndex == maxLimit) {
- break;
- }
- }
-
- if (log.isDebugEnabled()) {
- log.debug("Read lines " + fromIndex + " to " + (fromIndex + maxResults));
- }
-
- return resultList;
- }
-
- @Override
- public void initialize() {
- if (session != null || results != null) {
- close();
- }
-
- session = ((Session) entityManager.getDelegate()).getSessionFactory().openStatelessSession();
-
- this.prepareSQLSession();
-
- if (log.isDebugEnabled()) {
- if (maxResults >= 0) {
- log.debug("Preparing the query with " + maxResults + " results.");
- } else {
- log.debug("Preparing the query with all the results.");
- }
- }
-
- Query query = session.createSQLQuery(selectNativeQuery);
- if (maxResults >= 0) {
- query.setMaxResults(maxResults);
- }
- query.setFetchSize(30000);
- query.setReadOnly(true);
- query.setTimeout(TIMEOUT);
- results = query.scroll(ScrollMode.FORWARD_ONLY);
-
- lastMigratedItemIndex = -1;
-
- if (log.isDebugEnabled()) {
- if (maxResults >= 0) {
- log.debug("Query prepared with " + maxResults + " results.");
- } else {
- log.debug("Query prepared with all the results.");
- }
- }
- }
-
- @Override
- public void close() {
- if (results != null) {
- results.close();
- results = null;
- }
-
- if (session != null) {
- session.close();
- session = null;
- }
- }
-
- private void prepareSQLSession() {
- if (DatabaseType.Postgres.equals(this.databaseType)) {
- log.debug("Preparing SQL connection with timeout: " + DataMigrator.SQL_TIMEOUT);
-
- Query query = session.createSQLQuery("SET LOCAL statement_timeout = " + DataMigrator.SQL_TIMEOUT);
- query.setReadOnly(true);
- query.executeUpdate();
- }
- }
-}
diff --git a/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/Telemetry.java b/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/Telemetry.java
deleted file mode 100644
index 92bf37a..0000000
--- a/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/Telemetry.java
+++ /dev/null
@@ -1,57 +0,0 @@
-/*
- * RHQ Management Platform
- * Copyright 2013, Red Hat Middleware LLC, and individual contributors
- * as indicated by the @author tags. See the copyright.txt file in the
- * distribution for a full listing of individual contributors.
- *
- * This program is free software; you can redistribute it and/or modify
- * it under the terms of the GNU General Public License as published by
- * the Free Software Foundation version 2 of the License.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
- * GNU General Public License for more details.
- *
- * You should have received a copy of the GNU General Public License
- * along with this program; if not, write to the Free Software
- * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
- */
-
-package org.rhq.server.metrics.migrator;
-
-import org.apache.commons.lang.time.StopWatch;
-
-/**
- * @author Stefan Negrea
- *
- */
-public class Telemetry {
- private StopWatch generalTimer;
- private StopWatch migrationTimer;
-
- public Telemetry() {
- this.generalTimer = new StopWatch();
- this.migrationTimer = new StopWatch();
- }
-
- public StopWatch getGeneralTimer() {
- return generalTimer;
- }
-
- public StopWatch getMigrationTimer() {
- return migrationTimer;
- }
-
- public long getMigrationTime() {
- return migrationTimer.getTime();
- }
-
- public long getGeneralTime() {
- return generalTimer.getTime();
- }
-
- public long getNonMigrationTime() {
- return this.getGeneralTime() - this.getMigrationTime();
- }
-}
diff --git a/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/datasources/ExistingDataBulkExportSource.java b/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/datasources/ExistingDataBulkExportSource.java
new file mode 100644
index 0000000..82fd28f
--- /dev/null
+++ b/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/datasources/ExistingDataBulkExportSource.java
@@ -0,0 +1,124 @@
+/*
+ * RHQ Management Platform
+ * Copyright (C) 2005-2013 Red Hat, Inc.
+ * All rights reserved.
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation version 2 of the License.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software Foundation, Inc.,
+ * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA
+ */
+
+package org.rhq.server.metrics.migrator.datasources;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileReader;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Random;
+import java.util.StringTokenizer;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.rhq.core.util.stream.StreamUtil;
+
+/**
+ * @author Thomas Segismont
+ */
+public abstract class ExistingDataBulkExportSource implements ExistingDataSource {
+
+ private static final Log log = LogFactory.getLog(ExistingDataBulkExportSource.class);
+
+ protected static final int IO_BUFFER_SIZE = 1024 * 64;
+
+ protected static final String DELIMITER = "|";
+
+ private File existingDataFile;
+
+ private BufferedReader existingDataFileReader;
+
+ private int currentIndex;
+
+ public ExistingDataBulkExportSource() {
+ Random random = new Random();
+ String exportFileName = System.currentTimeMillis() + "." + random.nextInt() + ".export";
+ this.existingDataFile = new File(System.getProperty("java.io.tmpdir"), exportFileName);
+ }
+
+ @Override
+ public void initialize() throws Exception {
+ this.exportExistingData();
+ this.startReading();
+ }
+
+ @Override
+ public void close() throws Exception {
+ this.stopReading();
+ this.removeTempFile();
+ }
+
+ protected abstract void exportExistingData() throws Exception;
+
+ protected File getExistingDataFile() {
+ return existingDataFile;
+ }
+
+ protected void startReading() throws Exception {
+ if (!existingDataFile.exists() && !existingDataFile.isFile() && !existingDataFile.canRead()) {
+ throw new IllegalStateException();
+ }
+ existingDataFileReader = new BufferedReader(new FileReader(existingDataFile));
+ currentIndex = 0;
+ }
+
+ protected void stopReading() {
+ StreamUtil.safeClose(existingDataFileReader);
+ }
+
+ protected void removeTempFile(){
+ if(this.existingDataFile.exists()){
+ try{
+ this.existingDataFile.delete();
+ }catch(Exception e){
+ log.debug("Unable to clean temporary file " + this.existingDataFile, e);
+ }
+ }
+ }
+
+ @Override
+ public List<Object[]> getData(int fromIndex, int maxResults) throws Exception {
+ if (log.isDebugEnabled()) {
+ log.debug("Reading lines " + fromIndex + " to " + (fromIndex + maxResults));
+ }
+
+ if (fromIndex != currentIndex) {
+ throw new IllegalStateException();
+ }
+
+ List<Object[]> results = new LinkedList<Object[]>();
+ for (int i = 0; i < maxResults; i++) {
+ String nextLine = existingDataFileReader.readLine();
+ if (nextLine == null) {
+ break;
+ }
+ currentIndex++;
+ StringTokenizer stringTokenizer = new StringTokenizer(nextLine, DELIMITER);
+ Object[] row = new Object[stringTokenizer.countTokens()];
+ for (int j = 0; j < row.length; j++) {
+ row[j] = stringTokenizer.nextToken();
+ }
+ results.add(row);
+ }
+ return results;
+ }
+}
diff --git a/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/datasources/ExistingDataJPABulkExportSource.java b/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/datasources/ExistingDataJPABulkExportSource.java
new file mode 100644
index 0000000..f71c7b3
--- /dev/null
+++ b/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/datasources/ExistingDataJPABulkExportSource.java
@@ -0,0 +1,58 @@
+package org.rhq.server.metrics.migrator.datasources;
+
+import java.io.BufferedWriter;
+import java.io.FileWriter;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+
+import javax.persistence.EntityManager;
+
+import org.hibernate.Session;
+import org.hibernate.engine.spi.SessionFactoryImplementor;
+import org.hibernate.service.jdbc.connections.spi.ConnectionProvider;
+
+import org.rhq.core.util.jdbc.JDBCUtil;
+import org.rhq.core.util.stream.StreamUtil;
+
+/**
+ * @author Thomas Segismont
+ */
+public class ExistingDataJPABulkExportSource extends ExistingDataBulkExportSource {
+
+ private EntityManager entityManager;
+ private String selectNativeQuery;
+
+ public ExistingDataJPABulkExportSource(EntityManager entityManager, String selectNativeQuery) {
+ super();
+ }
+
+ protected void exportExistingData() throws Exception {
+ BufferedWriter fileWriter = null;
+ Connection connection = null;
+ PreparedStatement statement = null;
+ ResultSet resultSet = null;
+ try {
+ fileWriter = new BufferedWriter(new FileWriter(getExistingDataFile()), IO_BUFFER_SIZE);
+ Session session = (Session) this.entityManager.getDelegate();
+ SessionFactoryImplementor sfi = (SessionFactoryImplementor) session.getSessionFactory();
+ ConnectionProvider cp = sfi.getConnectionProvider();
+ connection = cp.getConnection();
+ statement = connection.prepareStatement(this.selectNativeQuery);
+ resultSet = statement.executeQuery();
+ int columnCount = resultSet.getMetaData().getColumnCount();
+ while (resultSet.next()) {
+ for (int i = 1; i < columnCount + 1; i++) {
+ if (i > 1) {
+ fileWriter.write(DELIMITER);
+ }
+ fileWriter.write(resultSet.getString(i));
+ }
+ fileWriter.write("\n");
+ }
+ } finally {
+ StreamUtil.safeClose(fileWriter);
+ JDBCUtil.safeClose(connection, statement, resultSet);
+ }
+ }
+}
diff --git a/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/datasources/ExistingDataJPASource.java b/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/datasources/ExistingDataJPASource.java
new file mode 100644
index 0000000..a14d8fb
--- /dev/null
+++ b/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/datasources/ExistingDataJPASource.java
@@ -0,0 +1,64 @@
+/*
+ * RHQ Management Platform
+ * Copyright (C) 2005-2013 Red Hat, Inc.
+ * All rights reserved.
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation version 2 of the License.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software Foundation, Inc.,
+ * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA
+ */
+
+package org.rhq.server.metrics.migrator.datasources;
+
+import java.util.List;
+
+import javax.persistence.EntityManager;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+/**
+ * @author Thomas Segismont
+ */
+public class ExistingDataJPASource implements ExistingDataSource {
+
+ private static final Log log = LogFactory.getLog(ExistingDataJPASource.class);
+
+ private EntityManager entityManager;
+ private String selectNativeQuery;
+
+ public ExistingDataJPASource(EntityManager entityManager, String selectNativeQuery) {
+ this.entityManager = entityManager;
+ this.selectNativeQuery = selectNativeQuery;
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public List<Object[]> getData(int fromIndex, int maxResults) throws Exception {
+ if (log.isDebugEnabled()) {
+ log.debug("Reading lines " + fromIndex + " to " + (fromIndex + maxResults));
+ }
+
+ return (List<Object[]>) entityManager.createNativeQuery(selectNativeQuery).setFirstResult(fromIndex)
+ .setMaxResults(maxResults).getResultList();
+ }
+
+ @Override
+ public void initialize() {
+ //nothing to do since it just implements a simple query with limits
+ }
+
+ @Override
+ public void close() {
+ //nothing to do since it just implements a simple query with limits
+ }
+}
diff --git a/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/datasources/ExistingDataSource.java b/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/datasources/ExistingDataSource.java
new file mode 100644
index 0000000..e6e4c95
--- /dev/null
+++ b/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/datasources/ExistingDataSource.java
@@ -0,0 +1,36 @@
+/*
+ * RHQ Management Platform
+ * Copyright (C) 2005-2013 Red Hat, Inc.
+ * All rights reserved.
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation version 2 of the License.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software Foundation, Inc.,
+ * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA
+ */
+
+package org.rhq.server.metrics.migrator.datasources;
+
+import java.util.List;
+
+/**
+ * @author Thomas Segismont
+ */
+public interface ExistingDataSource {
+
+ static final int TIMEOUT = 6000000;
+
+ List<Object[]> getData(int fromIndex, int maxResults) throws Exception;
+
+ void initialize() throws Exception;
+
+ void close() throws Exception;
+}
diff --git a/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/datasources/ExistingPostgresDataBulkExportSource.java b/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/datasources/ExistingPostgresDataBulkExportSource.java
new file mode 100644
index 0000000..ff08ca8
--- /dev/null
+++ b/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/datasources/ExistingPostgresDataBulkExportSource.java
@@ -0,0 +1,80 @@
+/*
+ * RHQ Management Platform
+ * Copyright (C) 2005-2013 Red Hat, Inc.
+ * All rights reserved.
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation version 2 of the License.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software Foundation, Inc.,
+ * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA
+ */
+
+package org.rhq.server.metrics.migrator.datasources;
+
+import java.io.BufferedWriter;
+import java.io.FileWriter;
+import java.sql.Connection;
+
+import javax.persistence.EntityManager;
+
+import org.hibernate.Session;
+import org.hibernate.engine.spi.SessionFactoryImplementor;
+import org.hibernate.service.jdbc.connections.spi.ConnectionProvider;
+import org.postgresql.copy.CopyManager;
+import org.postgresql.core.BaseConnection;
+
+import org.rhq.core.util.jdbc.JDBCUtil;
+import org.rhq.core.util.stream.StreamUtil;
+
+/**
+ * @author Thomas Segismont
+ */
+public class ExistingPostgresDataBulkExportSource extends ExistingDataBulkExportSource {
+
+ private final EntityManager entityManager;
+ private final String selectNativeQuery;
+ private final int maxResults;
+
+ public ExistingPostgresDataBulkExportSource(EntityManager entityManager, String selectNativeQuery) {
+ this(entityManager, selectNativeQuery, -1);
+ }
+
+ public ExistingPostgresDataBulkExportSource(EntityManager entityManager, String selectNativeQuery, int maxResults) {
+ super();
+ this.entityManager = entityManager;
+ this.selectNativeQuery = selectNativeQuery;
+ this.maxResults = maxResults;
+ }
+
+ protected void exportExistingData() throws Exception {
+ BufferedWriter fileWriter = null;
+ Connection connection = null;
+ try {
+ fileWriter = new BufferedWriter(new FileWriter(getExistingDataFile()), IO_BUFFER_SIZE);
+ Session session = (Session) entityManager.getDelegate();
+ SessionFactoryImplementor sfi = (SessionFactoryImplementor) session.getSessionFactory();
+ ConnectionProvider cp = sfi.getConnectionProvider();
+ connection = cp.getConnection();
+ CopyManager copyManager = new CopyManager((BaseConnection) connection);
+
+ if (maxResults > 0) {
+ copyManager.copyOut("COPY (" + selectNativeQuery + " LIMIT " + maxResults
+ + ") TO STDOUT WITH DELIMITER '" + DELIMITER + "'", fileWriter);
+ } else {
+ copyManager.copyOut("COPY (" + selectNativeQuery + ") TO STDOUT WITH DELIMITER '" + DELIMITER + "'",
+ fileWriter);
+ }
+ } finally {
+ StreamUtil.safeClose(fileWriter);
+ JDBCUtil.safeClose(connection);
+ }
+ }
+}
diff --git a/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/datasources/ScrollableDataSource.java b/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/datasources/ScrollableDataSource.java
new file mode 100644
index 0000000..a1792a6
--- /dev/null
+++ b/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/datasources/ScrollableDataSource.java
@@ -0,0 +1,157 @@
+/*
+ * RHQ Management Platform
+ * Copyright (C) 2005-2013 Red Hat, Inc.
+ * All rights reserved.
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation version 2 of the License.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software Foundation, Inc.,
+ * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA
+ */
+
+package org.rhq.server.metrics.migrator.datasources;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import javax.persistence.EntityManager;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.hibernate.Query;
+import org.hibernate.ScrollMode;
+import org.hibernate.ScrollableResults;
+import org.hibernate.Session;
+import org.hibernate.StatelessSession;
+
+import org.rhq.server.metrics.migrator.DataMigrator;
+import org.rhq.server.metrics.migrator.DataMigrator.DatabaseType;
+
+/**
+ * @author Stefan Negrea
+ */
+public class ScrollableDataSource implements ExistingDataSource {
+
+ private static final Log log = LogFactory.getLog(ScrollableDataSource.class);
+
+ private final EntityManager entityManager;
+ private final DatabaseType databaseType;
+ private final String selectNativeQuery;
+ private final int maxResults;
+
+ private ScrollableResults results;
+ private StatelessSession session;
+ private int lastMigratedItemIndex;
+
+
+ public ScrollableDataSource(EntityManager entityManager, DatabaseType databaseType,
+ String selectNativeQuery) {
+ this(entityManager, databaseType, selectNativeQuery, -1);
+ }
+
+ public ScrollableDataSource(EntityManager entityManager, DatabaseType databaseType, String selectNativeQuery,
+ int maxResults) {
+ this.entityManager = entityManager;
+ this.databaseType = databaseType;
+ this.selectNativeQuery = selectNativeQuery;
+ this.maxResults = maxResults;
+ }
+
+ @Override
+ public List<Object[]> getData(int fromIndex, int maxResults) throws Exception {
+ if (fromIndex != lastMigratedItemIndex + 1) {
+ throw new Exception("Cursor error. " + fromIndex + " " + lastMigratedItemIndex);
+ }
+
+ if (log.isDebugEnabled()) {
+ log.debug("Reading lines " + fromIndex + " to " + (fromIndex + maxResults));
+ }
+
+ List<Object[]> resultList = new ArrayList<Object[]>();
+
+ int maxLimit = lastMigratedItemIndex + maxResults;
+
+ while (results.next()) {
+ resultList.add(results.get());
+
+ if (++lastMigratedItemIndex == maxLimit) {
+ break;
+ }
+ }
+
+ if (log.isDebugEnabled()) {
+ log.debug("Read lines " + fromIndex + " to " + (fromIndex + maxResults));
+ }
+
+ return resultList;
+ }
+
+ @Override
+ public void initialize() {
+ if (session != null || results != null) {
+ close();
+ }
+
+ session = ((Session) entityManager.getDelegate()).getSessionFactory().openStatelessSession();
+
+ this.prepareSQLSession();
+
+ if (log.isDebugEnabled()) {
+ if (maxResults >= 0) {
+ log.debug("Preparing the query with " + maxResults + " results.");
+ } else {
+ log.debug("Preparing the query with all the results.");
+ }
+ }
+
+ Query query = session.createSQLQuery(selectNativeQuery);
+ if (maxResults >= 0) {
+ query.setMaxResults(maxResults);
+ }
+ query.setFetchSize(30000);
+ query.setReadOnly(true);
+ query.setTimeout(TIMEOUT);
+ results = query.scroll(ScrollMode.FORWARD_ONLY);
+
+ lastMigratedItemIndex = -1;
+
+ if (log.isDebugEnabled()) {
+ if (maxResults >= 0) {
+ log.debug("Query prepared with " + maxResults + " results.");
+ } else {
+ log.debug("Query prepared with all the results.");
+ }
+ }
+ }
+
+ @Override
+ public void close() {
+ if (results != null) {
+ results.close();
+ results = null;
+ }
+
+ if (session != null) {
+ session.close();
+ session = null;
+ }
+ }
+
+ private void prepareSQLSession() {
+ if (DatabaseType.Postgres.equals(this.databaseType)) {
+ log.debug("Preparing SQL connection with timeout: " + DataMigrator.SQL_TIMEOUT);
+
+ Query query = session.createSQLQuery("SET LOCAL statement_timeout = " + DataMigrator.SQL_TIMEOUT);
+ query.setReadOnly(true);
+ query.executeUpdate();
+ }
+ }
+}
diff --git a/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/workers/AbstractMigrationWorker.java b/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/workers/AbstractMigrationWorker.java
index 3c006be..c2ab85c 100644
--- a/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/workers/AbstractMigrationWorker.java
+++ b/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/workers/AbstractMigrationWorker.java
@@ -28,9 +28,9 @@ import org.rhq.server.metrics.migrator.DataMigrator;
import org.rhq.server.metrics.migrator.DataMigrator.DataMigratorConfiguration;
import org.rhq.server.metrics.migrator.DataMigrator.DatabaseType;
import org.rhq.server.metrics.migrator.DataMigrator.Task;
-import org.rhq.server.metrics.migrator.ExistingDataSource;
-import org.rhq.server.metrics.migrator.ExistingPostgresDataBulkExportSource;
-import org.rhq.server.metrics.migrator.ScrollableDataSource;
+import org.rhq.server.metrics.migrator.datasources.ExistingDataSource;
+import org.rhq.server.metrics.migrator.datasources.ExistingPostgresDataBulkExportSource;
+import org.rhq.server.metrics.migrator.datasources.ScrollableDataSource;
/**
* @author Stefan Negrea
diff --git a/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/workers/AggregateDataMigrator.java b/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/workers/AggregateDataMigrator.java
index 5511321..73116a4 100644
--- a/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/workers/AggregateDataMigrator.java
+++ b/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/workers/AggregateDataMigrator.java
@@ -39,8 +39,7 @@ import org.rhq.server.metrics.domain.MetricsTable;
import org.rhq.server.metrics.migrator.DataMigrator;
import org.rhq.server.metrics.migrator.DataMigrator.DataMigratorConfiguration;
import org.rhq.server.metrics.migrator.DataMigrator.Task;
-import org.rhq.server.metrics.migrator.ExistingDataSource;
-import org.rhq.server.metrics.migrator.Telemetry;
+import org.rhq.server.metrics.migrator.datasources.ExistingDataSource;
/**
* @author Stefan Negrea
diff --git a/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/workers/RawDataMigrator.java b/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/workers/RawDataMigrator.java
index 6680fd2..fd44a6f 100644
--- a/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/workers/RawDataMigrator.java
+++ b/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/workers/RawDataMigrator.java
@@ -41,8 +41,7 @@ import org.rhq.server.metrics.domain.MetricsTable;
import org.rhq.server.metrics.migrator.DataMigrator;
import org.rhq.server.metrics.migrator.DataMigrator.DataMigratorConfiguration;
import org.rhq.server.metrics.migrator.DataMigrator.Task;
-import org.rhq.server.metrics.migrator.ExistingDataSource;
-import org.rhq.server.metrics.migrator.Telemetry;
+import org.rhq.server.metrics.migrator.datasources.ExistingDataSource;
/**
* @author Stefan Negrea
@@ -125,7 +124,7 @@ public class RawDataMigrator extends AbstractMigrationWorker implements Callable
while (true) {
existingData = dataSource.getData(lastMigratedRecord, MAX_RECORDS_TO_LOAD_FROM_SQL);
- if (existingData.size() == 0) {
+ if (existingData == null || existingData.size() == 0) {
break;
}
diff --git a/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/workers/Telemetry.java b/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/workers/Telemetry.java
new file mode 100644
index 0000000..a16724b
--- /dev/null
+++ b/modules/enterprise/server/data-migration/src/main/java/org/rhq/server/metrics/migrator/workers/Telemetry.java
@@ -0,0 +1,57 @@
+/*
+ * RHQ Management Platform
+ * Copyright 2013, Red Hat Middleware LLC, and individual contributors
+ * as indicated by the @author tags. See the copyright.txt file in the
+ * distribution for a full listing of individual contributors.
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation version 2 of the License.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
+ */
+
+package org.rhq.server.metrics.migrator.workers;
+
+import org.apache.commons.lang.time.StopWatch;
+
+/**
+ * @author Stefan Negrea
+ *
+ */
+public class Telemetry {
+ private StopWatch generalTimer;
+ private StopWatch migrationTimer;
+
+ public Telemetry() {
+ this.generalTimer = new StopWatch();
+ this.migrationTimer = new StopWatch();
+ }
+
+ public StopWatch getGeneralTimer() {
+ return generalTimer;
+ }
+
+ public StopWatch getMigrationTimer() {
+ return migrationTimer;
+ }
+
+ public long getMigrationTime() {
+ return migrationTimer.getTime();
+ }
+
+ public long getGeneralTime() {
+ return generalTimer.getTime();
+ }
+
+ public long getNonMigrationTime() {
+ return this.getGeneralTime() - this.getMigrationTime();
+ }
+}
diff --git a/modules/enterprise/server/data-migration/src/test/java/org/rhq/server/metrics/DataSourceTest.java b/modules/enterprise/server/data-migration/src/test/java/org/rhq/server/metrics/DataSourceTest.java
deleted file mode 100644
index cda04ca..0000000
--- a/modules/enterprise/server/data-migration/src/test/java/org/rhq/server/metrics/DataSourceTest.java
+++ /dev/null
@@ -1,169 +0,0 @@
-/*
- * RHQ Management Platform
- * Copyright (C) 2005-2013 Red Hat, Inc.
- * All rights reserved.
- *
- * This program is free software; you can redistribute it and/or modify
- * it under the terms of the GNU General Public License as published by
- * the Free Software Foundation version 2 of the License.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
- * GNU General Public License for more details.
- *
- * You should have received a copy of the GNU General Public License
- * along with this program; if not, write to the Free Software Foundation, Inc.,
- * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA
- */
-
-package org.rhq.server.metrics;
-
-import java.util.List;
-import java.util.Properties;
-
-import javax.persistence.EntityManager;
-import javax.persistence.EntityManagerFactory;
-
-import org.apache.commons.lang.time.StopWatch;
-import org.apache.log4j.BasicConfigurator;
-import org.apache.log4j.Level;
-import org.apache.log4j.Logger;
-import org.hibernate.ejb.Ejb3Configuration;
-
-import org.rhq.server.metrics.migrator.ExistingDataBulkExportSource;
-import org.rhq.server.metrics.migrator.ExistingPostgresDataBulkExportSource;
-
-/**
- * @author Thomas Segismont
- */
-
-
-public class DataSourceTest {
-
- //ExistingPostgresDataBulkExport
- public static void main(String[] args) throws Exception {
- BasicConfigurator.configure();
- Logger.getRootLogger().setLevel(Level.INFO);
- Logger.getLogger("org.rhq").setLevel(Level.DEBUG);
- EntityManagerFactory entityManagerFactory = null;
- EntityManager entityManager = null;
- ExistingDataBulkExportSource source = null;
- try {
- entityManagerFactory = createEntityManager();
- entityManager = entityManagerFactory.createEntityManager();
- source = new ExistingPostgresDataBulkExportSource(entityManager,
- "SELECT schedule_id, time_stamp, value, minvalue, maxvalue FROM RHQ_MEASUREMENT_DATA_NUM_1D");
- StopWatch stopWatch = new StopWatch();
- stopWatch.start();
- source.initialize();
- int rowIndex = 0;
- int maxResults = 30000;
- for (;;) {
- List<Object[]> existingData = source.getData(rowIndex, maxResults);
- if (existingData.size() < maxResults) {
- break;
- } else {
- rowIndex += maxResults;
- }
- }
- stopWatch.stop();
- System.out.println("Execution: " + stopWatch);
- } finally {
- if (source != null) {
- source.close();
- }
- if (entityManager != null) {
- entityManager.close();
- }
- if (entityManagerFactory != null) {
- entityManagerFactory.close();
- }
- }
- }
-
- private static EntityManagerFactory createEntityManager() throws Exception {
- Properties properties = new Properties();
- properties.put("javax.persistence.provider", "org.hibernate.ejb.HibernatePersistence");
- properties.put("hibernate.connection.username", "rhqadmin");
- properties.put("hibernate.connection.password", "rhqadmin");
- properties.put("hibernate.dialect", "org.hibernate.dialect.PostgreSQLDialect");
- String driverClassName = "org.postgresql.Driver";
- try {
- //Required to preload the driver manually.
- //Without this the driver load will fail due to the packaging.
- Class.forName(driverClassName);
- } catch (ClassNotFoundException e) {
- throw new Exception("Postgres SQL Driver class could not be loaded. Missing class: " + driverClassName);
- }
- properties.put("hibernate.driver_class", driverClassName);
- properties.put("hibernate.connection.url", "jdbc:postgresql://localhost:5432/rhq");
- Ejb3Configuration configuration = new Ejb3Configuration();
- configuration.setProperties(properties);
- return configuration.buildEntityManagerFactory();
- }
-
- //ExistingDataJPABulkExportSource
-
- public static void main2(String[] args) throws Exception {
- BasicConfigurator.configure();
- Logger.getRootLogger().setLevel(Level.INFO);
- Logger.getLogger("org.rhq").setLevel(Level.DEBUG);
- EntityManagerFactory entityManagerFactory = null;
- EntityManager entityManager = null;
- ExistingDataBulkExportSource source = null;
- try {
- entityManagerFactory = createEntityManager();
- entityManager = entityManagerFactory.createEntityManager();
- source = new ExistingPostgresDataBulkExportSource(entityManager,
- "SELECT schedule_id, time_stamp, value, minvalue, maxvalue FROM RHQ_MEASUREMENT_DATA_NUM_1D");
- StopWatch stopWatch = new StopWatch();
- stopWatch.start();
- source.initialize();
- int rowIndex = 0;
- int maxResults = 30000;
- for (;;) {
- List<Object[]> existingData = source.getData(rowIndex, maxResults);
- if (existingData.size() < maxResults) {
- break;
- } else {
- rowIndex += maxResults;
- }
- }
- stopWatch.stop();
- System.out.println("Execution: " + stopWatch);
- } finally {
- if (source != null) {
- source.close();
- }
- if (entityManager != null) {
- entityManager.close();
- }
- if (entityManagerFactory != null) {
- entityManagerFactory.close();
- }
- }
- }
-
- private static EntityManagerFactory createEntityManager2() throws Exception {
- Properties properties = new Properties();
- properties.put("javax.persistence.provider", "org.hibernate.ejb.HibernatePersistence");
- properties.put("hibernate.connection.username", "rhqadmin");
- properties.put("hibernate.connection.password", "rhqadmin");
- properties.put("hibernate.dialect", "org.hibernate.dialect.PostgreSQLDialect");
- String driverClassName = "org.postgresql.Driver";
- try {
- //Required to preload the driver manually.
- //Without this the driver load will fail due to the packaging.
- Class.forName(driverClassName);
- } catch (ClassNotFoundException e) {
- throw new Exception("Postgres SQL Driver class could not be loaded. Missing class: " + driverClassName);
- }
- properties.put("hibernate.driver_class", driverClassName);
- properties.put("hibernate.connection.url", "jdbc:postgresql://localhost:5432/rhqdev");
- Ejb3Configuration configuration = new Ejb3Configuration();
- configuration.setProperties(properties);
- return configuration.buildEntityManagerFactory();
- }
-
-}
\ No newline at end of file
diff --git a/modules/enterprise/server/data-migration/src/test/java/org/rhq/server/metrics/migrator/DataMigratorTest.java b/modules/enterprise/server/data-migration/src/test/java/org/rhq/server/metrics/migrator/DataMigratorTest.java
new file mode 100644
index 0000000..71c6abe
--- /dev/null
+++ b/modules/enterprise/server/data-migration/src/test/java/org/rhq/server/metrics/migrator/DataMigratorTest.java
@@ -0,0 +1,235 @@
+/*
+ * RHQ Management Platform
+ * Copyright 2013, Red Hat Middleware LLC, and individual contributors
+ * as indicated by the @author tags. See the copyright.txt file in the
+ * distribution for a full listing of individual contributors.
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation version 2 of the License.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
+ */
+
+package org.rhq.server.metrics.migrator;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
+
+import javax.persistence.EntityManager;
+
+import com.datastax.driver.core.Session;
+
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.testng.PowerMockObjectFactory;
+import org.testng.Assert;
+import org.testng.IObjectFactory;
+import org.testng.annotations.ObjectFactory;
+import org.testng.annotations.Test;
+
+import org.rhq.server.metrics.migrator.DataMigrator.DataMigratorConfiguration;
+import org.rhq.server.metrics.migrator.DataMigrator.DatabaseType;
+import org.rhq.server.metrics.migrator.workers.AggregateDataMigrator;
+import org.rhq.server.metrics.migrator.workers.RawDataMigrator;
+
+@PrepareForTest({ DataMigrator.class, DataMigratorConfiguration.class, RawDataMigrator.class })
+public class DataMigratorTest {
+
+ @ObjectFactory
+ public IObjectFactory getObjectFactory() {
+ return new PowerMockObjectFactory();
+ }
+
+ @Test
+ public void testSetup() throws Exception {
+ //tell the method story as it happens: mock or create dependencies and configure
+ //those dependencies to get the method under test to completion
+ DatabaseType databaseType = DatabaseType.Postgres;
+ EntityManager mockEntityManager = mock(EntityManager.class);
+ Session mockCassandraSession = mock(Session.class);
+
+ DataMigratorConfiguration mockConfig = PowerMockito.mock(DataMigratorConfiguration.class);
+ PowerMockito.whenNew(DataMigratorConfiguration.class)
+ .withArguments(eq(mockEntityManager), eq(mockCassandraSession), eq(databaseType), eq(false)).thenReturn(mockConfig);
+
+ //create object to test and inject required dependencies
+ DataMigrator objectUnderTest = new DataMigrator(mockEntityManager, mockCassandraSession, databaseType);
+
+ //run code under test
+ objectUnderTest.runRawDataMigration(false);
+ objectUnderTest.runRawDataMigration(true);
+
+ objectUnderTest.run1HAggregateDataMigration(false);
+ objectUnderTest.run1HAggregateDataMigration(true);
+
+ objectUnderTest.run6HAggregateDataMigration(false);
+ objectUnderTest.run6HAggregateDataMigration(true);
+
+ objectUnderTest.run1DAggregateDataMigration(false);
+ objectUnderTest.run1DAggregateDataMigration(true);
+
+
+ //verify the results (assert and mock verification)
+ PowerMockito.verifyNew(DataMigratorConfiguration.class).withArguments(eq(mockEntityManager), eq(mockCassandraSession),
+ eq(databaseType), eq(false));
+ PowerMockito.verifyPrivate(mockConfig, times(2)).invoke("setRunRawDataMigration", true);
+ PowerMockito.verifyPrivate(mockConfig, times(1)).invoke("setRunRawDataMigration", false);
+
+ PowerMockito.verifyPrivate(mockConfig, times(2)).invoke("setRun1HAggregateDataMigration", true);
+ PowerMockito.verifyPrivate(mockConfig, times(1)).invoke("setRun1HAggregateDataMigration", false);
+
+ PowerMockito.verifyPrivate(mockConfig, times(2)).invoke("setRun6HAggregateDataMigration", true);
+ PowerMockito.verifyPrivate(mockConfig, times(1)).invoke("setRun6HAggregateDataMigration", false);
+
+ PowerMockito.verifyPrivate(mockConfig, times(2)).invoke("setRun1DAggregateDataMigration", true);
+ PowerMockito.verifyPrivate(mockConfig, times(1)).invoke("setRun1DAggregateDataMigration", false);
+
+ PowerMockito.verifyPrivate(mockConfig, times(1)).invoke("setDeleteDataImmediatelyAfterMigration", false);
+ PowerMockito.verifyPrivate(mockConfig, times(1)).invoke("setDeleteAllDataAtEndOfMigration", false);
+
+ PowerMockito.verifyNoMoreInteractions(mockConfig);
+
+ verifyNoMoreInteractions(mockEntityManager);
+ verifyNoMoreInteractions(mockCassandraSession);
+ }
+
+ @Test
+ public void testEstimateTask() throws Exception {
+ //tell the method story as it happens: mock or create dependencies and configure
+ //those dependencies to get the method under test to completion
+ DatabaseType databaseType = DatabaseType.Postgres;
+
+ EntityManager mockEntityManager = mock(EntityManager.class);
+ Session mockCassandraSession = mock(Session.class);
+ DataMigratorConfiguration mockConfig = PowerMockito.mock(DataMigratorConfiguration.class);
+ PowerMockito.whenNew(DataMigratorConfiguration.class)
+ .withArguments(eq(mockEntityManager), eq(mockCassandraSession), eq(databaseType), eq(false)).thenReturn(mockConfig);
+
+ when(mockConfig.isRunRawDataMigration()).thenReturn(true);
+ when(mockConfig.isRun1HAggregateDataMigration()).thenReturn(false);
+ when(mockConfig.isRun6HAggregateDataMigration()).thenReturn(false);
+ when(mockConfig.isRun1DAggregateDataMigration()).thenReturn(false);
+
+ RawDataMigrator mockRawDataMigrator = mock(RawDataMigrator.class);
+ PowerMockito.whenNew(RawDataMigrator.class).withArguments(eq(mockConfig)).thenReturn(mockRawDataMigrator);
+ long estimateExpected = 1234L;
+ when(mockRawDataMigrator.estimate()).thenReturn(estimateExpected);
+
+ //create object to test and inject required dependencies
+ DataMigrator objectUnderTest = new DataMigrator(mockEntityManager, mockCassandraSession, databaseType);
+
+ //run code under test
+ objectUnderTest.run1HAggregateDataMigration(false);
+ objectUnderTest.run6HAggregateDataMigration(false);
+ objectUnderTest.run1DAggregateDataMigration(false);
+
+ long estimateActual = objectUnderTest.estimate();
+
+ //verify the results (assert and mock verification)
+ PowerMockito.verifyNew(DataMigratorConfiguration.class).withArguments(eq(mockEntityManager), eq(mockCassandraSession),
+ eq(databaseType), eq(false));
+
+ PowerMockito.verifyPrivate(mockConfig, times(1)).invoke("setRunRawDataMigration", true);
+
+ PowerMockito.verifyPrivate(mockConfig, times(1)).invoke("setRun1HAggregateDataMigration", true);
+ PowerMockito.verifyPrivate(mockConfig, times(1)).invoke("setRun1HAggregateDataMigration", false);
+
+ PowerMockito.verifyPrivate(mockConfig, times(1)).invoke("setRun6HAggregateDataMigration", true);
+ PowerMockito.verifyPrivate(mockConfig, times(1)).invoke("setRun6HAggregateDataMigration", false);
+
+ PowerMockito.verifyPrivate(mockConfig, times(1)).invoke("setRun1DAggregateDataMigration", true);
+ PowerMockito.verifyPrivate(mockConfig, times(1)).invoke("setRun1DAggregateDataMigration", false);
+
+ PowerMockito.verifyPrivate(mockConfig, times(1)).invoke("setDeleteDataImmediatelyAfterMigration", false);
+ PowerMockito.verifyPrivate(mockConfig, times(1)).invoke("setDeleteAllDataAtEndOfMigration", false);
+
+ verify(mockConfig, times(1)).isRunRawDataMigration();
+ verify(mockConfig, times(1)).isRun1HAggregateDataMigration();
+ verify(mockConfig, times(1)).isRun6HAggregateDataMigration();
+ verify(mockConfig, times(1)).isRun1DAggregateDataMigration();
+ verify(mockConfig, times(1)).isDeleteAllDataAtEndOfMigration();
+
+ PowerMockito.verifyNoMoreInteractions(mockConfig);
+
+ PowerMockito.verifyNew(RawDataMigrator.class).withArguments(eq(mockConfig));
+ verify(mockRawDataMigrator, times(1)).estimate();
+ verifyNoMoreInteractions(mockRawDataMigrator);
+
+ verifyNoMoreInteractions(mockEntityManager);
+ verifyNoMoreInteractions(mockCassandraSession);
+
+ Assert.assertEquals(estimateActual, (long) (estimateExpected + estimateExpected * .15));
+ }
+
+ @Test
+ public void testMigrateTask() throws Exception {
+ //tell the method story as it happens: mock or create dependencies and configure
+ //those dependencies to get the method under test to completion
+ DatabaseType databaseType = DatabaseType.Postgres;
+
+ EntityManager mockEntityManager = mock(EntityManager.class);
+ Session mockCassandraSession = mock(Session.class);
+ DataMigratorConfiguration mockConfig = PowerMockito.mock(DataMigratorConfiguration.class);
+ PowerMockito.whenNew(DataMigratorConfiguration.class)
+ .withArguments(eq(mockEntityManager), eq(mockCassandraSession), eq(databaseType), eq(false)).thenReturn(mockConfig);
+
+ when(mockConfig.isRunRawDataMigration()).thenReturn(false);
+ when(mockConfig.isRun1HAggregateDataMigration()).thenReturn(true);
+ when(mockConfig.isRun6HAggregateDataMigration()).thenReturn(true);
+ when(mockConfig.isRun1DAggregateDataMigration()).thenReturn(true);
+
+ AggregateDataMigrator mockAggregateDataMigrator = mock(AggregateDataMigrator.class);
+ PowerMockito.whenNew(AggregateDataMigrator.class).withArguments(any(), eq(mockConfig))
+ .thenReturn(mockAggregateDataMigrator);
+
+ //create object to test and inject required dependencies
+ DataMigrator objectUnderTest = new DataMigrator(mockEntityManager, mockCassandraSession, databaseType);
+
+ //run code under test
+ objectUnderTest.runRawDataMigration(false);
+ objectUnderTest.migrateData();
+
+ //verify the results (assert and mock verification)
+ PowerMockito.verifyNew(DataMigratorConfiguration.class).withArguments(eq(mockEntityManager), eq(mockCassandraSession),
+ eq(databaseType), eq(false));
+
+ PowerMockito.verifyPrivate(mockConfig, times(1)).invoke("setRunRawDataMigration", true);
+ PowerMockito.verifyPrivate(mockConfig, times(1)).invoke("setRunRawDataMigration", false);
+
+ PowerMockito.verifyPrivate(mockConfig, times(1)).invoke("setRun1HAggregateDataMigration", true);
+ PowerMockito.verifyPrivate(mockConfig, times(1)).invoke("setRun6HAggregateDataMigration", true);
+ PowerMockito.verifyPrivate(mockConfig, times(1)).invoke("setRun1DAggregateDataMigration", true);
+
+ PowerMockito.verifyPrivate(mockConfig, times(1)).invoke("setDeleteDataImmediatelyAfterMigration", false);
+ PowerMockito.verifyPrivate(mockConfig, times(1)).invoke("setDeleteAllDataAtEndOfMigration", false);
+
+ verify(mockConfig, times(1)).isRunRawDataMigration();
+ verify(mockConfig, times(1)).isRun1HAggregateDataMigration();
+ verify(mockConfig, times(1)).isRun6HAggregateDataMigration();
+ verify(mockConfig, times(1)).isRun1DAggregateDataMigration();
+ verify(mockConfig, times(1)).isDeleteAllDataAtEndOfMigration();
+
+ PowerMockito.verifyNoMoreInteractions(mockConfig);
+
+ PowerMockito.verifyNew(AggregateDataMigrator.class, times(3)).withArguments(any(), eq(mockConfig));
+ verify(mockAggregateDataMigrator, times(3)).migrate();
+ verifyNoMoreInteractions(mockAggregateDataMigrator);
+
+ verifyNoMoreInteractions(mockEntityManager);
+ verifyNoMoreInteractions(mockCassandraSession);
+ }
+}
+
diff --git a/modules/enterprise/server/data-migration/src/test/java/org/rhq/server/metrics/migrator/DataSourceTest.java b/modules/enterprise/server/data-migration/src/test/java/org/rhq/server/metrics/migrator/DataSourceTest.java
new file mode 100644
index 0000000..3112f33
--- /dev/null
+++ b/modules/enterprise/server/data-migration/src/test/java/org/rhq/server/metrics/migrator/DataSourceTest.java
@@ -0,0 +1,169 @@
+/*
+ * RHQ Management Platform
+ * Copyright (C) 2005-2013 Red Hat, Inc.
+ * All rights reserved.
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation version 2 of the License.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software Foundation, Inc.,
+ * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA
+ */
+
+package org.rhq.server.metrics.migrator;
+
+import java.util.List;
+import java.util.Properties;
+
+import javax.persistence.EntityManager;
+import javax.persistence.EntityManagerFactory;
+
+import org.apache.commons.lang.time.StopWatch;
+import org.apache.log4j.BasicConfigurator;
+import org.apache.log4j.Level;
+import org.apache.log4j.Logger;
+import org.hibernate.ejb.Ejb3Configuration;
+
+import org.rhq.server.metrics.migrator.datasources.ExistingDataBulkExportSource;
+import org.rhq.server.metrics.migrator.datasources.ExistingPostgresDataBulkExportSource;
+
+/**
+ * @author Thomas Segismont
+ */
+
+
+public class DataSourceTest {
+
+ //ExistingPostgresDataBulkExport
+ public static void main(String[] args) throws Exception {
+ BasicConfigurator.configure();
+ Logger.getRootLogger().setLevel(Level.INFO);
+ Logger.getLogger("org.rhq").setLevel(Level.DEBUG);
+ EntityManagerFactory entityManagerFactory = null;
+ EntityManager entityManager = null;
+ ExistingDataBulkExportSource source = null;
+ try {
+ entityManagerFactory = createEntityManager();
+ entityManager = entityManagerFactory.createEntityManager();
+ source = new ExistingPostgresDataBulkExportSource(entityManager,
+ "SELECT schedule_id, time_stamp, value, minvalue, maxvalue FROM RHQ_MEASUREMENT_DATA_NUM_1D");
+ StopWatch stopWatch = new StopWatch();
+ stopWatch.start();
+ source.initialize();
+ int rowIndex = 0;
+ int maxResults = 30000;
+ for (;;) {
+ List<Object[]> existingData = source.getData(rowIndex, maxResults);
+ if (existingData.size() < maxResults) {
+ break;
+ } else {
+ rowIndex += maxResults;
+ }
+ }
+ stopWatch.stop();
+ System.out.println("Execution: " + stopWatch);
+ } finally {
+ if (source != null) {
+ source.close();
+ }
+ if (entityManager != null) {
+ entityManager.close();
+ }
+ if (entityManagerFactory != null) {
+ entityManagerFactory.close();
+ }
+ }
+ }
+
+ private static EntityManagerFactory createEntityManager() throws Exception {
+ Properties properties = new Properties();
+ properties.put("javax.persistence.provider", "org.hibernate.ejb.HibernatePersistence");
+ properties.put("hibernate.connection.username", "rhqadmin");
+ properties.put("hibernate.connection.password", "rhqadmin");
+ properties.put("hibernate.dialect", "org.hibernate.dialect.PostgreSQLDialect");
+ String driverClassName = "org.postgresql.Driver";
+ try {
+ //Required to preload the driver manually.
+ //Without this the driver load will fail due to the packaging.
+ Class.forName(driverClassName);
+ } catch (ClassNotFoundException e) {
+ throw new Exception("Postgres SQL Driver class could not be loaded. Missing class: " + driverClassName);
+ }
+ properties.put("hibernate.driver_class", driverClassName);
+ properties.put("hibernate.connection.url", "jdbc:postgresql://localhost:5432/rhq");
+ Ejb3Configuration configuration = new Ejb3Configuration();
+ configuration.setProperties(properties);
+ return configuration.buildEntityManagerFactory();
+ }
+
+ //ExistingDataJPABulkExportSource
+
+ public static void main2(String[] args) throws Exception {
+ BasicConfigurator.configure();
+ Logger.getRootLogger().setLevel(Level.INFO);
+ Logger.getLogger("org.rhq").setLevel(Level.DEBUG);
+ EntityManagerFactory entityManagerFactory = null;
+ EntityManager entityManager = null;
+ ExistingDataBulkExportSource source = null;
+ try {
+ entityManagerFactory = createEntityManager();
+ entityManager = entityManagerFactory.createEntityManager();
+ source = new ExistingPostgresDataBulkExportSource(entityManager,
+ "SELECT schedule_id, time_stamp, value, minvalue, maxvalue FROM RHQ_MEASUREMENT_DATA_NUM_1D");
+ StopWatch stopWatch = new StopWatch();
+ stopWatch.start();
+ source.initialize();
+ int rowIndex = 0;
+ int maxResults = 30000;
+ for (;;) {
+ List<Object[]> existingData = source.getData(rowIndex, maxResults);
+ if (existingData.size() < maxResults) {
+ break;
+ } else {
+ rowIndex += maxResults;
+ }
+ }
+ stopWatch.stop();
+ System.out.println("Execution: " + stopWatch);
+ } finally {
+ if (source != null) {
+ source.close();
+ }
+ if (entityManager != null) {
+ entityManager.close();
+ }
+ if (entityManagerFactory != null) {
+ entityManagerFactory.close();
+ }
+ }
+ }
+
+ private static EntityManagerFactory createEntityManager2() throws Exception {
+ Properties properties = new Properties();
+ properties.put("javax.persistence.provider", "org.hibernate.ejb.HibernatePersistence");
+ properties.put("hibernate.connection.username", "rhqadmin");
+ properties.put("hibernate.connection.password", "rhqadmin");
+ properties.put("hibernate.dialect", "org.hibernate.dialect.PostgreSQLDialect");
+ String driverClassName = "org.postgresql.Driver";
+ try {
+ //Required to preload the driver manually.
+ //Without this the driver load will fail due to the packaging.
+ Class.forName(driverClassName);
+ } catch (ClassNotFoundException e) {
+ throw new Exception("Postgres SQL Driver class could not be loaded. Missing class: " + driverClassName);
+ }
+ properties.put("hibernate.driver_class", driverClassName);
+ properties.put("hibernate.connection.url", "jdbc:postgresql://localhost:5432/rhqdev");
+ Ejb3Configuration configuration = new Ejb3Configuration();
+ configuration.setProperties(properties);
+ return configuration.buildEntityManagerFactory();
+ }
+
+}
\ No newline at end of file
diff --git a/modules/enterprise/server/data-migration/src/test/java/org/rhq/server/metrics/migrator/workers/AggregateDataMigratorTest.java b/modules/enterprise/server/data-migration/src/test/java/org/rhq/server/metrics/migrator/workers/AggregateDataMigratorTest.java
new file mode 100644
index 0000000..0a29a4b
--- /dev/null
+++ b/modules/enterprise/server/data-migration/src/test/java/org/rhq/server/metrics/migrator/workers/AggregateDataMigratorTest.java
@@ -0,0 +1,198 @@
+/*
+ * RHQ Management Platform
+ * Copyright 2013, Red Hat Middleware LLC, and individual contributors
+ * as indicated by the @author tags. See the copyright.txt file in the
+ * distribution for a full listing of individual contributors.
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation version 2 of the License.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
+ */
+
+package org.rhq.server.metrics.migrator.workers;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyInt;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import javax.persistence.EntityManager;
+
+import com.datastax.driver.core.Query;
+import com.datastax.driver.core.ResultSetFuture;
+import com.datastax.driver.core.Session;
+
+import org.hibernate.SessionFactory;
+import org.hibernate.StatelessSession;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.testng.PowerMockObjectFactory;
+import org.testng.Assert;
+import org.testng.IObjectFactory;
+import org.testng.annotations.ObjectFactory;
+import org.testng.annotations.Test;
+
+import org.rhq.server.metrics.domain.MetricsTable;
+import org.rhq.server.metrics.migrator.DataMigrator.DataMigratorConfiguration;
+import org.rhq.server.metrics.migrator.DataMigrator.DatabaseType;
+import org.rhq.server.metrics.migrator.datasources.ScrollableDataSource;
+
+
+@PrepareForTest({ AggregateDataMigrator.class })
+public class AggregateDataMigratorTest {
+
+ @ObjectFactory
+ public IObjectFactory getObjectFactory() {
+ return new PowerMockObjectFactory();
+ }
+
+ @Test
+ public void testEstimateTask() throws Exception {
+ //tell the method story as it happens: mock or create dependencies and configure
+ //those dependencies to get the method under test to completion
+ DatabaseType databaseType = DatabaseType.Oracle;
+ DataMigratorConfiguration mockConfig = mock(DataMigratorConfiguration.class);
+ when(mockConfig.getDatabaseType()).thenReturn(databaseType);
+
+ MetricsIndexUpdateAccumulator mockMetricsIndexUpdateAccumulator = mock(MetricsIndexUpdateAccumulator.class);
+ PowerMockito.whenNew(MetricsIndexUpdateAccumulator.class)
+ .withArguments(eq(MetricsTable.ONE_HOUR), eq(mockConfig))
+ .thenReturn(mockMetricsIndexUpdateAccumulator);
+
+ EntityManager mockEntityManager = mock(EntityManager.class);
+ when(mockConfig.getEntityManager()).thenReturn(mockEntityManager);
+
+ org.hibernate.Session mockHibernateSession = mock(org.hibernate.Session.class);
+ when(mockEntityManager.getDelegate()).thenReturn(mockHibernateSession);
+ SessionFactory mockSessionFactory = mock(SessionFactory.class);
+ when(mockHibernateSession.getSessionFactory()).thenReturn(mockSessionFactory);
+ StatelessSession mockStatelessSession = mock(StatelessSession.class);
+ when(mockSessionFactory.openStatelessSession()).thenReturn(mockStatelessSession);
+
+ org.hibernate.SQLQuery mockQuery = mock(org.hibernate.SQLQuery.class);
+ when(mockStatelessSession.createSQLQuery(any(String.class))).thenReturn(mockQuery);
+
+ when(mockQuery.uniqueResult()).thenReturn("1000");
+
+ ScrollableDataSource mockDataSource = mock(ScrollableDataSource.class);
+ PowerMockito.whenNew(ScrollableDataSource.class)
+ .withArguments(eq(mockEntityManager), eq(databaseType), any(), anyInt()).thenReturn(mockDataSource);
+ when(mockDataSource.getData(eq(0), anyInt())).thenReturn(new ArrayList<Object[]>());
+
+ //create object to test and inject required dependencies
+ AggregateDataMigrator objectUnderTest = new AggregateDataMigrator(MetricsTable.ONE_HOUR, mockConfig);
+
+ //run code under test
+ long estimateActual = objectUnderTest.estimate();
+
+ //verify the results (assert and mock verification)
+ PowerMockito.verifyNew(MetricsIndexUpdateAccumulator.class).withArguments(eq(MetricsTable.ONE_HOUR),
+ eq(mockConfig));
+ PowerMockito.verifyNew(ScrollableDataSource.class, times(1)).withArguments(eq(mockEntityManager),
+ eq(databaseType), any(), anyInt());
+
+ verify(mockStatelessSession, times(1)).createSQLQuery(any(String.class));
+ verify(mockDataSource, times(1)).initialize();
+ verify(mockDataSource, times(1)).getData(eq(0), anyInt());
+ verify(mockDataSource, times(1)).close();
+ verify(mockMetricsIndexUpdateAccumulator, times(1)).drain();
+
+ verifyNoMoreInteractions(mockDataSource);
+ verifyNoMoreInteractions(mockMetricsIndexUpdateAccumulator);
+
+ Assert.assertNotEquals(estimateActual, 0);
+ }
+
+ @Test
+ public void testMigrateTask() throws Exception {
+ //tell the method story as it happens: mock or create dependencies and configure
+ //those dependencies to get the method under test to completion
+ DatabaseType databaseType = DatabaseType.Oracle;
+ DataMigratorConfiguration mockConfig = mock(DataMigratorConfiguration.class);
+ when(mockConfig.getDatabaseType()).thenReturn(databaseType);
+
+ Session mockCassandraSession = mock(Session.class);
+ when(mockConfig.getSession()).thenReturn(mockCassandraSession);
+
+ MetricsIndexUpdateAccumulator mockMetricsIndexUpdateAccumulator = mock(MetricsIndexUpdateAccumulator.class);
+ PowerMockito.whenNew(MetricsIndexUpdateAccumulator.class)
+ .withArguments(eq(MetricsTable.ONE_HOUR), eq(mockConfig))
+ .thenReturn(mockMetricsIndexUpdateAccumulator);
+
+ EntityManager mockEntityManager = mock(EntityManager.class);
+ when(mockConfig.getEntityManager()).thenReturn(mockEntityManager);
+
+ org.hibernate.Session mockHibernateSession = mock(org.hibernate.Session.class);
+ when(mockEntityManager.getDelegate()).thenReturn(mockHibernateSession);
+ SessionFactory mockSessionFactory = mock(SessionFactory.class);
+ when(mockHibernateSession.getSessionFactory()).thenReturn(mockSessionFactory);
+ StatelessSession mockStatelessSession = mock(StatelessSession.class);
+ when(mockSessionFactory.openStatelessSession()).thenReturn(mockStatelessSession);
+
+ org.hibernate.SQLQuery mockQuery = mock(org.hibernate.SQLQuery.class);
+ when(mockStatelessSession.createSQLQuery(any(String.class))).thenReturn(mockQuery);
+
+ when(mockQuery.uniqueResult()).thenReturn("1000");
+
+ ScrollableDataSource mockDataSource = mock(ScrollableDataSource.class);
+ PowerMockito.whenNew(ScrollableDataSource.class).withArguments(eq(mockEntityManager), eq(databaseType), any())
+ .thenReturn(mockDataSource);
+
+ List<Object[]> resultList = new ArrayList<Object[]>();
+ resultList.add(new Object[] { 100, 100, 100, 100, 100 });
+ resultList.add(new Object[] { 100, System.currentTimeMillis() - 100l, 100, 100, 100 });
+
+ for (int index = 0; index < 15; index++) {
+ when(mockDataSource.getData(eq(0), anyInt())).thenReturn(resultList);
+ when(mockDataSource.getData(eq(2), anyInt())).thenReturn(new ArrayList<Object[]>());
+ }
+
+ ResultSetFuture mockResultSetFuture = mock(ResultSetFuture.class);
+ when(mockCassandraSession.executeAsync(any(Query.class))).thenReturn(mockResultSetFuture);
+
+ //create object to test and inject required dependencies
+ AggregateDataMigrator objectUnderTest = new AggregateDataMigrator(MetricsTable.ONE_HOUR, mockConfig);
+
+ //run code under test
+ objectUnderTest.migrate();
+
+ //verify the results (assert and mock verification)
+ PowerMockito.verifyNew(MetricsIndexUpdateAccumulator.class).withArguments(eq(MetricsTable.ONE_HOUR),
+ eq(mockConfig));
+ PowerMockito.verifyNew(ScrollableDataSource.class, times(1)).withArguments(eq(mockEntityManager),
+ eq(databaseType), any());
+
+ verify(mockDataSource, times(1)).initialize();
+ verify(mockDataSource, times(1)).getData(eq(0), anyInt());
+ verify(mockDataSource, times(1)).getData(eq(2), anyInt());
+ verify(mockDataSource, times(1)).close();
+
+ verify(mockMetricsIndexUpdateAccumulator, times(1)).add(eq(100), anyInt());
+ verify(mockMetricsIndexUpdateAccumulator, times(1)).drain();
+
+ verify(mockCassandraSession, times(1)).executeAsync(any(Query.class));
+ verify(mockResultSetFuture, times(1)).get();
+
+ verifyNoMoreInteractions(mockDataSource);
+ verifyNoMoreInteractions(mockCassandraSession);
+ verifyNoMoreInteractions(mockResultSetFuture);
+ verifyNoMoreInteractions(mockMetricsIndexUpdateAccumulator);
+ }
+}
+
diff --git a/modules/enterprise/server/data-migration/src/test/java/org/rhq/server/metrics/migrator/workers/RawDataMigratorTest.java b/modules/enterprise/server/data-migration/src/test/java/org/rhq/server/metrics/migrator/workers/RawDataMigratorTest.java
new file mode 100644
index 0000000..6bf35e0
--- /dev/null
+++ b/modules/enterprise/server/data-migration/src/test/java/org/rhq/server/metrics/migrator/workers/RawDataMigratorTest.java
@@ -0,0 +1,194 @@
+/*
+ * RHQ Management Platform
+ * Copyright 2013, Red Hat Middleware LLC, and individual contributors
+ * as indicated by the @author tags. See the copyright.txt file in the
+ * distribution for a full listing of individual contributors.
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation version 2 of the License.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
+ */
+
+package org.rhq.server.metrics.migrator.workers;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyInt;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import javax.persistence.EntityManager;
+
+import com.datastax.driver.core.Query;
+import com.datastax.driver.core.ResultSetFuture;
+import com.datastax.driver.core.Session;
+
+import org.hibernate.SessionFactory;
+import org.hibernate.StatelessSession;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.testng.PowerMockObjectFactory;
+import org.testng.Assert;
+import org.testng.IObjectFactory;
+import org.testng.annotations.ObjectFactory;
+import org.testng.annotations.Test;
+
+import org.rhq.server.metrics.domain.MetricsTable;
+import org.rhq.server.metrics.migrator.DataMigrator.DataMigratorConfiguration;
+import org.rhq.server.metrics.migrator.DataMigrator.DatabaseType;
+import org.rhq.server.metrics.migrator.datasources.ScrollableDataSource;
+
+
+@PrepareForTest({ RawDataMigrator.class })
+public class RawDataMigratorTest {
+
+ @ObjectFactory
+ public IObjectFactory getObjectFactory() {
+ return new PowerMockObjectFactory();
+ }
+
+ @Test
+ public void testEstimateTask() throws Exception {
+ //tell the method story as it happens: mock or create dependencies and configure
+ //those dependencies to get the method under test to completion
+ DatabaseType databaseType = DatabaseType.Oracle;
+ DataMigratorConfiguration mockConfig = mock(DataMigratorConfiguration.class);
+ when(mockConfig.getDatabaseType()).thenReturn(databaseType);
+
+ MetricsIndexUpdateAccumulator mockMetricsIndexUpdateAccumulator = mock(MetricsIndexUpdateAccumulator.class);
+ PowerMockito.whenNew(MetricsIndexUpdateAccumulator.class).withArguments(eq(MetricsTable.RAW), eq(mockConfig))
+ .thenReturn(mockMetricsIndexUpdateAccumulator);
+
+ EntityManager mockEntityManager = mock(EntityManager.class);
+ when(mockConfig.getEntityManager()).thenReturn(mockEntityManager);
+
+ org.hibernate.Session mockHibernateSession = mock(org.hibernate.Session.class);
+ when(mockEntityManager.getDelegate()).thenReturn(mockHibernateSession);
+ SessionFactory mockSessionFactory = mock(SessionFactory.class);
+ when(mockHibernateSession.getSessionFactory()).thenReturn(mockSessionFactory);
+ StatelessSession mockStatelessSession = mock(StatelessSession.class);
+ when(mockSessionFactory.openStatelessSession()).thenReturn(mockStatelessSession);
+
+ org.hibernate.SQLQuery mockQuery = mock(org.hibernate.SQLQuery.class);
+ when(mockStatelessSession.createSQLQuery(any(String.class))).thenReturn(mockQuery);
+
+ when(mockQuery.uniqueResult()).thenReturn("1000");
+
+ ScrollableDataSource mockDataSource = mock(ScrollableDataSource.class);
+ PowerMockito.whenNew(ScrollableDataSource.class)
+ .withArguments(eq(mockEntityManager), eq(databaseType), any(), anyInt()).thenReturn(mockDataSource);
+ when(mockDataSource.getData(eq(0), anyInt())).thenReturn(new ArrayList<Object[]>());
+
+ //create object to test and inject required dependencies
+ RawDataMigrator objectUnderTest = new RawDataMigrator(mockConfig);
+
+ //run code under test
+ long estimateActual = objectUnderTest.estimate();
+
+ //verify the results (assert and mock verification)
+ PowerMockito.verifyNew(MetricsIndexUpdateAccumulator.class).withArguments(eq(MetricsTable.RAW), eq(mockConfig));
+ PowerMockito.verifyNew(ScrollableDataSource.class, times(15)).withArguments(eq(mockEntityManager),
+ eq(databaseType), any(), anyInt());
+
+ verify(mockStatelessSession, times(15)).createSQLQuery(any(String.class));
+ verify(mockDataSource, times(15)).initialize();
+ verify(mockDataSource, times(15)).getData(eq(0), anyInt());
+ verify(mockDataSource, times(15)).close();
+ verify(mockMetricsIndexUpdateAccumulator, times(1)).drain();
+
+ verifyNoMoreInteractions(mockDataSource);
+ verifyNoMoreInteractions(mockMetricsIndexUpdateAccumulator);
+
+ Assert.assertNotEquals(estimateActual, 0);
+ }
+
+ @Test
+ public void testMigrateTask() throws Exception {
+ //tell the method story as it happens: mock or create dependencies and configure
+ //those dependencies to get the method under test to completion
+ DatabaseType databaseType = DatabaseType.Oracle;
+ DataMigratorConfiguration mockConfig = mock(DataMigratorConfiguration.class);
+ when(mockConfig.getDatabaseType()).thenReturn(databaseType);
+
+ Session mockCassandraSession = mock(Session.class);
+ when(mockConfig.getSession()).thenReturn(mockCassandraSession);
+
+ MetricsIndexUpdateAccumulator mockMetricsIndexUpdateAccumulator = mock(MetricsIndexUpdateAccumulator.class);
+ PowerMockito.whenNew(MetricsIndexUpdateAccumulator.class).withArguments(eq(MetricsTable.RAW), eq(mockConfig))
+ .thenReturn(mockMetricsIndexUpdateAccumulator);
+
+ EntityManager mockEntityManager = mock(EntityManager.class);
+ when(mockConfig.getEntityManager()).thenReturn(mockEntityManager);
+
+ org.hibernate.Session mockHibernateSession = mock(org.hibernate.Session.class);
+ when(mockEntityManager.getDelegate()).thenReturn(mockHibernateSession);
+ SessionFactory mockSessionFactory = mock(SessionFactory.class);
+ when(mockHibernateSession.getSessionFactory()).thenReturn(mockSessionFactory);
+ StatelessSession mockStatelessSession = mock(StatelessSession.class);
+ when(mockSessionFactory.openStatelessSession()).thenReturn(mockStatelessSession);
+
+ org.hibernate.SQLQuery mockQuery = mock(org.hibernate.SQLQuery.class);
+ when(mockStatelessSession.createSQLQuery(any(String.class))).thenReturn(mockQuery);
+
+ when(mockQuery.uniqueResult()).thenReturn("1000");
+
+ ScrollableDataSource mockDataSource = mock(ScrollableDataSource.class);
+ PowerMockito.whenNew(ScrollableDataSource.class).withArguments(eq(mockEntityManager), eq(databaseType), any())
+ .thenReturn(mockDataSource);
+
+ List<Object[]> resultList = new ArrayList<Object[]>();
+ resultList.add(new Object[] { 100, 100, 100 });
+ resultList.add(new Object[] { 100, System.currentTimeMillis() - 100l, 100 });
+
+ for (int index = 0; index < 15; index++) {
+ when(mockDataSource.getData(eq(0), anyInt())).thenReturn(resultList);
+ when(mockDataSource.getData(eq(2), anyInt())).thenReturn(new ArrayList<Object[]>());
+ }
+
+ ResultSetFuture mockResultSetFuture = mock(ResultSetFuture.class);
+ when(mockCassandraSession.executeAsync(any(Query.class))).thenReturn(mockResultSetFuture);
+
+ //create object to test and inject required dependencies
+ RawDataMigrator objectUnderTest = new RawDataMigrator(mockConfig);
+
+ //run code under test
+ objectUnderTest.migrate();
+
+ //verify the results (assert and mock verification)
+ PowerMockito.verifyNew(MetricsIndexUpdateAccumulator.class).withArguments(eq(MetricsTable.RAW), eq(mockConfig));
+ PowerMockito.verifyNew(ScrollableDataSource.class, times(15)).withArguments(eq(mockEntityManager),
+ eq(databaseType), any());
+
+ verify(mockDataSource, times(15)).initialize();
+ verify(mockDataSource, times(15)).getData(eq(0), anyInt());
+ verify(mockDataSource, times(15)).getData(eq(2), anyInt());
+ verify(mockDataSource, times(15)).close();
+
+ verify(mockMetricsIndexUpdateAccumulator, times(15)).add(eq(100), anyInt());
+ verify(mockMetricsIndexUpdateAccumulator, times(1)).drain();
+
+ verify(mockCassandraSession, times(15)).executeAsync(any(Query.class));
+ verify(mockResultSetFuture, times(15)).get();
+
+ verifyNoMoreInteractions(mockDataSource);
+ verifyNoMoreInteractions(mockCassandraSession);
+ verifyNoMoreInteractions(mockResultSetFuture);
+ verifyNoMoreInteractions(mockMetricsIndexUpdateAccumulator);
+ }
+}
+