You are viewing a plain text version of this content. The canonical link for it is here.
Posted to server-dev@james.apache.org by bt...@apache.org on 2017/08/01 11:03:23 UTC
james-project git commit: JAMES-2096 Limit the number of
simultaneusly fetched messages during migration
Repository: james-project
Updated Branches:
refs/heads/master ea823c645 -> 7c333a50d
JAMES-2096 Limit the number of simultaneusly fetched messages during migration
Project: http://git-wip-us.apache.org/repos/asf/james-project/repo
Commit: http://git-wip-us.apache.org/repos/asf/james-project/commit/7c333a50
Tree: http://git-wip-us.apache.org/repos/asf/james-project/tree/7c333a50
Diff: http://git-wip-us.apache.org/repos/asf/james-project/diff/7c333a50
Branch: refs/heads/master
Commit: 7c333a50d85ff43ed64a5931063abf35e7dfeccb
Parents: ea823c6
Author: benwa <bt...@linagora.com>
Authored: Tue Aug 1 15:11:53 2017 +0700
Committer: benwa <bt...@linagora.com>
Committed: Tue Aug 1 16:04:48 2017 +0700
----------------------------------------------------------------------
.../cassandra/mail/CassandraMessageDAO.java | 17 ++++++++++-------
.../cassandra/mail/migration/V1ToV2Migration.java | 15 ++++++++++++++-
2 files changed, 24 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/james-project/blob/7c333a50/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageDAO.java
----------------------------------------------------------------------
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageDAO.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageDAO.java
index ba4802e..1c73063 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageDAO.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageDAO.java
@@ -97,7 +97,7 @@ public class CassandraMessageDAO {
private final PreparedStatement selectHeaders;
private final PreparedStatement selectFields;
private final PreparedStatement selectBody;
- private final PreparedStatement selectAll;
+ private final PreparedStatement selectByBatch;
private CassandraUtils cassandraUtils;
private final CassandraConfiguration cassandraConfiguration;
@@ -113,7 +113,7 @@ public class CassandraMessageDAO {
this.selectFields = prepareSelect(session, FIELDS);
this.selectBody = prepareSelect(session, BODY);
this.cassandraConfiguration = cassandraConfiguration;
- this.selectAll = prepareSelectAll(session);
+ this.selectByBatch = prepareSelectBatch(session, cassandraConfiguration);
this.cassandraUtils = cassandraUtils;
}
@@ -122,8 +122,9 @@ public class CassandraMessageDAO {
this(session, typesProvider, CassandraConfiguration.DEFAULT_CONFIGURATION, CassandraUtils.WITH_DEFAULT_CONFIGURATION);
}
- private PreparedStatement prepareSelectAll(Session session) {
- return session.prepare(select().from(TABLE_NAME));
+ private PreparedStatement prepareSelectBatch(Session session, CassandraConfiguration cassandraConfiguration) {
+ return session.prepare(select().from(TABLE_NAME)
+ .limit(cassandraConfiguration.getFetchNextPageInAdvanceRow()));
}
private PreparedStatement prepareSelect(Session session, String[] fields) {
@@ -152,11 +153,13 @@ public class CassandraMessageDAO {
.where(eq(MESSAGE_ID, bindMarker(MESSAGE_ID))));
}
- public Stream<RawMessage> readAll() {
+ public List<RawMessage> readBatch() {
return cassandraUtils.convertToStream(
- cassandraAsyncExecutor.execute(selectAll.bind().setFetchSize(cassandraConfiguration.getV1ReadFetchSize()))
+ cassandraAsyncExecutor.execute(selectByBatch.bind()
+ .setFetchSize(cassandraConfiguration.getV1ReadFetchSize()))
.join())
- .map(this::fromRow);
+ .map(this::fromRow)
+ .collect(Guavate.toImmutableList());
}
public CompletableFuture<Void> save(MailboxMessage message) throws MailboxException {
http://git-wip-us.apache.org/repos/asf/james-project/blob/7c333a50/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/migration/V1ToV2Migration.java
----------------------------------------------------------------------
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/migration/V1ToV2Migration.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/migration/V1ToV2Migration.java
index 3d50f84..db87065 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/migration/V1ToV2Migration.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/migration/V1ToV2Migration.java
@@ -116,7 +116,20 @@ public class V1ToV2Migration implements Migration {
@Override
public MigrationResult run() {
- return messageDAOV1.readAll()
+ boolean allResultFetched = false;
+ MigrationResult result = MigrationResult.COMPLETED;
+
+ while (!allResultFetched) {
+ List<CassandraMessageDAO.RawMessage> batch = messageDAOV1.readBatch();
+ allResultFetched = batch.isEmpty();
+ result = Migration.combine(result, migrateBatch(batch));
+ }
+ return result;
+ }
+
+ private MigrationResult migrateBatch(List<CassandraMessageDAO.RawMessage> batch) {
+ return batch
+ .stream()
.map(this::migrate)
.reduce(MigrationResult.COMPLETED, Migration::combine);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org