You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by pa...@apache.org on 2017/09/12 13:38:24 UTC
[1/6] cassandra git commit: Fix pending view mutations handling and
cleanup batchlog when there are local and remote paired mutations
Repository: cassandra
Updated Branches:
refs/heads/cassandra-3.0 e86bef439 -> 12103653f
refs/heads/cassandra-3.11 b64a4e4f6 -> eb027a1de
refs/heads/trunk 471835815 -> 826c9f4a6
Fix pending view mutations handling and cleanup batchlog when there are local and remote paired mutations
Patch by Paulo Motta; Reviewed by Sylvain Lebresne for CASSANDRA-13069
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/12103653
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/12103653
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/12103653
Branch: refs/heads/cassandra-3.0
Commit: 12103653f313d6f1ef030a535986123ddcffea9c
Parents: e86bef4
Author: Paulo Motta <pa...@gmail.com>
Authored: Wed Dec 21 20:19:21 2016 -0200
Committer: Paulo Motta <pa...@apache.org>
Committed: Tue Sep 12 08:30:24 2017 -0500
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../cassandra/batchlog/BatchlogManager.java | 3 +-
.../service/BatchlogResponseHandler.java | 4 +-
.../apache/cassandra/service/StorageProxy.java | 82 +++++++++-----------
4 files changed, 41 insertions(+), 49 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/12103653/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 76d155e..26b1794 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
3.0.15
+ * Fix pending view mutations handling and cleanup batchlog when there are local and remote paired mutations (CASSANDRA-13069)
* Range deletes in a CAS batch are ignored (CASSANDRA-13655)
* Change repair midpoint logging for tiny ranges (CASSANDRA-13603)
* Better handle corrupt final commitlog segment (CASSANDRA-11995)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/12103653/src/java/org/apache/cassandra/batchlog/BatchlogManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/batchlog/BatchlogManager.java b/src/java/org/apache/cassandra/batchlog/BatchlogManager.java
index b614fc5..a0b614f 100644
--- a/src/java/org/apache/cassandra/batchlog/BatchlogManager.java
+++ b/src/java/org/apache/cassandra/batchlog/BatchlogManager.java
@@ -67,6 +67,7 @@ public class BatchlogManager implements BatchlogManagerMBean
private static final Logger logger = LoggerFactory.getLogger(BatchlogManager.class);
public static final BatchlogManager instance = new BatchlogManager();
+ public static final long BATCHLOG_REPLAY_TIMEOUT = Long.getLong("cassandra.batchlog.replay_timeout_in_ms", DatabaseDescriptor.getWriteRpcTimeout() * 2);
private volatile long totalBatchesReplayed = 0; // no concurrency protection necessary as only written by replay thread.
private volatile UUID lastReplayedUuid = UUIDGen.minTimeUUID(0);
@@ -284,7 +285,7 @@ public class BatchlogManager implements BatchlogManagerMBean
public static long getBatchlogTimeout()
{
- return DatabaseDescriptor.getWriteRpcTimeout() * 2; // enough time for the actual write + BM removal mutation
+ return BATCHLOG_REPLAY_TIMEOUT; // enough time for the actual write + BM removal mutation
}
private static class ReplayingBatch
http://git-wip-us.apache.org/repos/asf/cassandra/blob/12103653/src/java/org/apache/cassandra/service/BatchlogResponseHandler.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/BatchlogResponseHandler.java b/src/java/org/apache/cassandra/service/BatchlogResponseHandler.java
index ac44923..a1477e6 100644
--- a/src/java/org/apache/cassandra/service/BatchlogResponseHandler.java
+++ b/src/java/org/apache/cassandra/service/BatchlogResponseHandler.java
@@ -50,7 +50,7 @@ public class BatchlogResponseHandler<T> extends AbstractWriteResponseHandler<T>
{
wrapped.response(msg);
if (requiredBeforeFinishUpdater.decrementAndGet(this) == 0)
- cleanup.run();
+ cleanup.ackMutation();
}
public boolean isLatencyForSnitch()
@@ -107,7 +107,7 @@ public class BatchlogResponseHandler<T> extends AbstractWriteResponseHandler<T>
this.callback = callback;
}
- public void run()
+ public void ackMutation()
{
if (mutationsWaitingForUpdater.decrementAndGet(this) == 0)
callback.invoke();
http://git-wip-us.apache.org/repos/asf/cassandra/blob/12103653/src/java/org/apache/cassandra/service/StorageProxy.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java
index 6610cf7..1ce1bc5 100644
--- a/src/java/org/apache/cassandra/service/StorageProxy.java
+++ b/src/java/org/apache/cassandra/service/StorageProxy.java
@@ -729,15 +729,16 @@ public class StorageProxy implements StorageProxyMBean
else
{
List<WriteResponseHandlerWrapper> wrappers = new ArrayList<>(mutations.size());
- List<Mutation> nonPairedMutations = new LinkedList<>();
+ //non-local mutations rely on the base mutation commit-log entry for eventual consistency
+ Set<Mutation> nonLocalMutations = new HashSet<>(mutations);
Token baseToken = StorageService.instance.getTokenMetadata().partitioner.getToken(dataKey);
ConsistencyLevel consistencyLevel = ConsistencyLevel.ONE;
//Since the base -> view replication is 1:1 we only need to store the BL locally
final Collection<InetAddress> batchlogEndpoints = Collections.singleton(FBUtilities.getBroadcastAddress());
- BatchlogResponseHandler.BatchlogCleanup cleanup = new BatchlogResponseHandler.BatchlogCleanup(mutations.size(),
- () -> asyncRemoveFromBatchlog(batchlogEndpoints, batchUUID));
+ BatchlogResponseHandler.BatchlogCleanup cleanup = new BatchlogResponseHandler.BatchlogCleanup(mutations.size(), () -> asyncRemoveFromBatchlog(batchlogEndpoints, batchUUID));
+
// add a handler for each mutation - includes checking availability, but doesn't initiate any writes, yet
for (Mutation mutation : mutations)
{
@@ -746,61 +747,50 @@ public class StorageProxy implements StorageProxyMBean
Optional<InetAddress> pairedEndpoint = ViewUtils.getViewNaturalEndpoint(keyspaceName, baseToken, tk);
Collection<InetAddress> pendingEndpoints = StorageService.instance.getTokenMetadata().pendingEndpointsFor(tk, keyspaceName);
- if (pairedEndpoint.isPresent())
- {
- // When local node is the endpoint and there are no pending nodes we can
- // Just apply the mutation locally.
- if (pairedEndpoint.get().equals(FBUtilities.getBroadcastAddress())
- && pendingEndpoints.isEmpty() && StorageService.instance.isJoined())
- try
- {
- mutation.apply(writeCommitLog);
- }
- catch (Exception exc)
- {
- logger.error("Error applying local view update to keyspace {}: {}", mutation.getKeyspaceName(), mutation);
- throw exc;
- }
- else
- {
- wrappers.add(wrapViewBatchResponseHandler(mutation,
- consistencyLevel,
- consistencyLevel,
- Collections.singletonList(pairedEndpoint.get()),
- baseComplete,
- WriteType.BATCH,
- cleanup));
- }
- }
- else
+ // if there are no paired endpoints there are probably range movements going on, so we write to the local batchlog to replay later
+ if (!pairedEndpoint.isPresent())
{
- //if there are no paired endpoints there are probably range movements going on,
- //so we write to the local batchlog to replay later
if (pendingEndpoints.isEmpty())
logger.warn("Received base materialized view mutation for key {} that does not belong " +
"to this node. There is probably a range movement happening (move or decommission)," +
"but this node hasn't updated its ring metadata yet. Adding mutation to " +
"local batchlog to be replayed later.",
mutation.key());
- nonPairedMutations.add(mutation);
+ continue;
+ }
+
+ // When local node is the paired endpoint just apply the mutation locally.
+ if (pairedEndpoint.get().equals(FBUtilities.getBroadcastAddress()) && StorageService.instance.isJoined())
+ try
+ {
+ mutation.apply(writeCommitLog);
+ nonLocalMutations.remove(mutation);
+ cleanup.ackMutation();
+ }
+ catch (Exception exc)
+ {
+ logger.error("Error applying local view update to keyspace {}: {}", mutation.getKeyspaceName(), mutation);
+ throw exc;
+ }
+ else
+ {
+ wrappers.add(wrapViewBatchResponseHandler(mutation,
+ consistencyLevel,
+ consistencyLevel,
+ Collections.singletonList(pairedEndpoint.get()),
+ baseComplete,
+ WriteType.BATCH,
+ cleanup));
}
}
- if (!wrappers.isEmpty())
- {
- // Apply to local batchlog memtable in this thread
- BatchlogManager.store(Batch.createLocal(batchUUID, FBUtilities.timestampMicros(), Lists.transform(wrappers, w -> w.mutation)),
- writeCommitLog);
+ // Apply to local batchlog memtable in this thread
+ if (!nonLocalMutations.isEmpty())
+ BatchlogManager.store(Batch.createLocal(batchUUID, FBUtilities.timestampMicros(), nonLocalMutations), writeCommitLog);
- // now actually perform the writes and wait for them to complete
+ // Perform remote writes
+ if (!wrappers.isEmpty())
asyncWriteBatchedMutations(wrappers, localDataCenter, Stage.VIEW_MUTATION);
- }
-
- if (!nonPairedMutations.isEmpty())
- {
- BatchlogManager.store(Batch.createLocal(batchUUID, FBUtilities.timestampMicros(), nonPairedMutations),
- writeCommitLog);
- }
}
}
finally
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
[6/6] cassandra git commit: Merge branch 'cassandra-3.11' into trunk
Posted by pa...@apache.org.
Merge branch 'cassandra-3.11' into trunk
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/826c9f4a
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/826c9f4a
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/826c9f4a
Branch: refs/heads/trunk
Commit: 826c9f4a6ebad8880390f8a26058d0c1f964f687
Parents: 4718358 eb027a1
Author: Paulo Motta <pa...@apache.org>
Authored: Tue Sep 12 08:33:19 2017 -0500
Committer: Paulo Motta <pa...@apache.org>
Committed: Tue Sep 12 08:33:19 2017 -0500
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../cassandra/batchlog/BatchlogManager.java | 3 +-
.../service/BatchlogResponseHandler.java | 4 +-
.../apache/cassandra/service/StorageProxy.java | 86 +++++++++-----------
4 files changed, 44 insertions(+), 50 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/826c9f4a/CHANGES.txt
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/826c9f4a/src/java/org/apache/cassandra/batchlog/BatchlogManager.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/826c9f4a/src/java/org/apache/cassandra/service/BatchlogResponseHandler.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/826c9f4a/src/java/org/apache/cassandra/service/StorageProxy.java
----------------------------------------------------------------------
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
[5/6] cassandra git commit: Merge branch 'cassandra-3.0' into
cassandra-3.11
Posted by pa...@apache.org.
Merge branch 'cassandra-3.0' into cassandra-3.11
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/eb027a1d
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/eb027a1d
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/eb027a1d
Branch: refs/heads/cassandra-3.11
Commit: eb027a1de6d97f040ef9da1552e5811539e633a0
Parents: b64a4e4 1210365
Author: Paulo Motta <pa...@apache.org>
Authored: Tue Sep 12 08:31:42 2017 -0500
Committer: Paulo Motta <pa...@apache.org>
Committed: Tue Sep 12 08:33:00 2017 -0500
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../cassandra/batchlog/BatchlogManager.java | 3 +-
.../service/BatchlogResponseHandler.java | 4 +-
.../apache/cassandra/service/StorageProxy.java | 86 +++++++++-----------
4 files changed, 44 insertions(+), 50 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/eb027a1d/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index 52775e7,26b1794..099a869
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -1,16 -1,6 +1,17 @@@
-3.0.15
+3.11.1
+ * Add a compaction option to TWCS to ignore sstables overlapping checks (CASSANDRA-13418)
+ * BTree.Builder memory leak (CASSANDRA-13754)
+ * Revert CASSANDRA-10368 of supporting non-pk column filtering due to correctness (CASSANDRA-13798)
+ * Fix cassandra-stress hang issues when an error during cluster connection happens (CASSANDRA-12938)
+ * Better bootstrap failure message when blocked by (potential) range movement (CASSANDRA-13744)
+ * "ignore" option is ignored in sstableloader (CASSANDRA-13721)
+ * Deadlock in AbstractCommitLogSegmentManager (CASSANDRA-13652)
+ * Duplicate the buffer before passing it to analyser in SASI operation (CASSANDRA-13512)
+ * Properly evict pstmts from prepared statements cache (CASSANDRA-13641)
+Merged from 3.0:
+ * Fix pending view mutations handling and cleanup batchlog when there are local and remote paired mutations (CASSANDRA-13069)
* Range deletes in a CAS batch are ignored (CASSANDRA-13655)
+ * Avoid assertion error when IndexSummary > 2G (CASSANDRA-12014)
* Change repair midpoint logging for tiny ranges (CASSANDRA-13603)
* Better handle corrupt final commitlog segment (CASSANDRA-11995)
* StreamingHistogram is not thread safe (CASSANDRA-13756)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/eb027a1d/src/java/org/apache/cassandra/batchlog/BatchlogManager.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/eb027a1d/src/java/org/apache/cassandra/service/BatchlogResponseHandler.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/eb027a1d/src/java/org/apache/cassandra/service/StorageProxy.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/service/StorageProxy.java
index b8f87d9,1ce1bc5..913c3c2
--- a/src/java/org/apache/cassandra/service/StorageProxy.java
+++ b/src/java/org/apache/cassandra/service/StorageProxy.java
@@@ -819,7 -756,31 +792,34 @@@ public class StorageProxy implements St
"but this node hasn't updated its ring metadata yet. Adding mutation to " +
"local batchlog to be replayed later.",
mutation.key());
- nonPairedMutations.add(mutation);
+ continue;
+ }
+
+ // When local node is the paired endpoint just apply the mutation locally.
+ if (pairedEndpoint.get().equals(FBUtilities.getBroadcastAddress()) && StorageService.instance.isJoined())
++ {
+ try
+ {
+ mutation.apply(writeCommitLog);
+ nonLocalMutations.remove(mutation);
+ cleanup.ackMutation();
+ }
+ catch (Exception exc)
+ {
+ logger.error("Error applying local view update to keyspace {}: {}", mutation.getKeyspaceName(), mutation);
+ throw exc;
+ }
++ }
+ else
+ {
+ wrappers.add(wrapViewBatchResponseHandler(mutation,
+ consistencyLevel,
+ consistencyLevel,
+ Collections.singletonList(pairedEndpoint.get()),
+ baseComplete,
+ WriteType.BATCH,
- cleanup));
++ cleanup,
++ queryStartNanoTime));
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
[3/6] cassandra git commit: Fix pending view mutations handling and
cleanup batchlog when there are local and remote paired mutations
Posted by pa...@apache.org.
Fix pending view mutations handling and cleanup batchlog when there are local and remote paired mutations
Patch by Paulo Motta; Reviewed by Sylvain Lebresne for CASSANDRA-13069
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/12103653
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/12103653
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/12103653
Branch: refs/heads/trunk
Commit: 12103653f313d6f1ef030a535986123ddcffea9c
Parents: e86bef4
Author: Paulo Motta <pa...@gmail.com>
Authored: Wed Dec 21 20:19:21 2016 -0200
Committer: Paulo Motta <pa...@apache.org>
Committed: Tue Sep 12 08:30:24 2017 -0500
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../cassandra/batchlog/BatchlogManager.java | 3 +-
.../service/BatchlogResponseHandler.java | 4 +-
.../apache/cassandra/service/StorageProxy.java | 82 +++++++++-----------
4 files changed, 41 insertions(+), 49 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/12103653/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 76d155e..26b1794 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
3.0.15
+ * Fix pending view mutations handling and cleanup batchlog when there are local and remote paired mutations (CASSANDRA-13069)
* Range deletes in a CAS batch are ignored (CASSANDRA-13655)
* Change repair midpoint logging for tiny ranges (CASSANDRA-13603)
* Better handle corrupt final commitlog segment (CASSANDRA-11995)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/12103653/src/java/org/apache/cassandra/batchlog/BatchlogManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/batchlog/BatchlogManager.java b/src/java/org/apache/cassandra/batchlog/BatchlogManager.java
index b614fc5..a0b614f 100644
--- a/src/java/org/apache/cassandra/batchlog/BatchlogManager.java
+++ b/src/java/org/apache/cassandra/batchlog/BatchlogManager.java
@@ -67,6 +67,7 @@ public class BatchlogManager implements BatchlogManagerMBean
private static final Logger logger = LoggerFactory.getLogger(BatchlogManager.class);
public static final BatchlogManager instance = new BatchlogManager();
+ public static final long BATCHLOG_REPLAY_TIMEOUT = Long.getLong("cassandra.batchlog.replay_timeout_in_ms", DatabaseDescriptor.getWriteRpcTimeout() * 2);
private volatile long totalBatchesReplayed = 0; // no concurrency protection necessary as only written by replay thread.
private volatile UUID lastReplayedUuid = UUIDGen.minTimeUUID(0);
@@ -284,7 +285,7 @@ public class BatchlogManager implements BatchlogManagerMBean
public static long getBatchlogTimeout()
{
- return DatabaseDescriptor.getWriteRpcTimeout() * 2; // enough time for the actual write + BM removal mutation
+ return BATCHLOG_REPLAY_TIMEOUT; // enough time for the actual write + BM removal mutation
}
private static class ReplayingBatch
http://git-wip-us.apache.org/repos/asf/cassandra/blob/12103653/src/java/org/apache/cassandra/service/BatchlogResponseHandler.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/BatchlogResponseHandler.java b/src/java/org/apache/cassandra/service/BatchlogResponseHandler.java
index ac44923..a1477e6 100644
--- a/src/java/org/apache/cassandra/service/BatchlogResponseHandler.java
+++ b/src/java/org/apache/cassandra/service/BatchlogResponseHandler.java
@@ -50,7 +50,7 @@ public class BatchlogResponseHandler<T> extends AbstractWriteResponseHandler<T>
{
wrapped.response(msg);
if (requiredBeforeFinishUpdater.decrementAndGet(this) == 0)
- cleanup.run();
+ cleanup.ackMutation();
}
public boolean isLatencyForSnitch()
@@ -107,7 +107,7 @@ public class BatchlogResponseHandler<T> extends AbstractWriteResponseHandler<T>
this.callback = callback;
}
- public void run()
+ public void ackMutation()
{
if (mutationsWaitingForUpdater.decrementAndGet(this) == 0)
callback.invoke();
http://git-wip-us.apache.org/repos/asf/cassandra/blob/12103653/src/java/org/apache/cassandra/service/StorageProxy.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java
index 6610cf7..1ce1bc5 100644
--- a/src/java/org/apache/cassandra/service/StorageProxy.java
+++ b/src/java/org/apache/cassandra/service/StorageProxy.java
@@ -729,15 +729,16 @@ public class StorageProxy implements StorageProxyMBean
else
{
List<WriteResponseHandlerWrapper> wrappers = new ArrayList<>(mutations.size());
- List<Mutation> nonPairedMutations = new LinkedList<>();
+ //non-local mutations rely on the base mutation commit-log entry for eventual consistency
+ Set<Mutation> nonLocalMutations = new HashSet<>(mutations);
Token baseToken = StorageService.instance.getTokenMetadata().partitioner.getToken(dataKey);
ConsistencyLevel consistencyLevel = ConsistencyLevel.ONE;
//Since the base -> view replication is 1:1 we only need to store the BL locally
final Collection<InetAddress> batchlogEndpoints = Collections.singleton(FBUtilities.getBroadcastAddress());
- BatchlogResponseHandler.BatchlogCleanup cleanup = new BatchlogResponseHandler.BatchlogCleanup(mutations.size(),
- () -> asyncRemoveFromBatchlog(batchlogEndpoints, batchUUID));
+ BatchlogResponseHandler.BatchlogCleanup cleanup = new BatchlogResponseHandler.BatchlogCleanup(mutations.size(), () -> asyncRemoveFromBatchlog(batchlogEndpoints, batchUUID));
+
// add a handler for each mutation - includes checking availability, but doesn't initiate any writes, yet
for (Mutation mutation : mutations)
{
@@ -746,61 +747,50 @@ public class StorageProxy implements StorageProxyMBean
Optional<InetAddress> pairedEndpoint = ViewUtils.getViewNaturalEndpoint(keyspaceName, baseToken, tk);
Collection<InetAddress> pendingEndpoints = StorageService.instance.getTokenMetadata().pendingEndpointsFor(tk, keyspaceName);
- if (pairedEndpoint.isPresent())
- {
- // When local node is the endpoint and there are no pending nodes we can
- // Just apply the mutation locally.
- if (pairedEndpoint.get().equals(FBUtilities.getBroadcastAddress())
- && pendingEndpoints.isEmpty() && StorageService.instance.isJoined())
- try
- {
- mutation.apply(writeCommitLog);
- }
- catch (Exception exc)
- {
- logger.error("Error applying local view update to keyspace {}: {}", mutation.getKeyspaceName(), mutation);
- throw exc;
- }
- else
- {
- wrappers.add(wrapViewBatchResponseHandler(mutation,
- consistencyLevel,
- consistencyLevel,
- Collections.singletonList(pairedEndpoint.get()),
- baseComplete,
- WriteType.BATCH,
- cleanup));
- }
- }
- else
+ // if there are no paired endpoints there are probably range movements going on, so we write to the local batchlog to replay later
+ if (!pairedEndpoint.isPresent())
{
- //if there are no paired endpoints there are probably range movements going on,
- //so we write to the local batchlog to replay later
if (pendingEndpoints.isEmpty())
logger.warn("Received base materialized view mutation for key {} that does not belong " +
"to this node. There is probably a range movement happening (move or decommission)," +
"but this node hasn't updated its ring metadata yet. Adding mutation to " +
"local batchlog to be replayed later.",
mutation.key());
- nonPairedMutations.add(mutation);
+ continue;
+ }
+
+ // When local node is the paired endpoint just apply the mutation locally.
+ if (pairedEndpoint.get().equals(FBUtilities.getBroadcastAddress()) && StorageService.instance.isJoined())
+ try
+ {
+ mutation.apply(writeCommitLog);
+ nonLocalMutations.remove(mutation);
+ cleanup.ackMutation();
+ }
+ catch (Exception exc)
+ {
+ logger.error("Error applying local view update to keyspace {}: {}", mutation.getKeyspaceName(), mutation);
+ throw exc;
+ }
+ else
+ {
+ wrappers.add(wrapViewBatchResponseHandler(mutation,
+ consistencyLevel,
+ consistencyLevel,
+ Collections.singletonList(pairedEndpoint.get()),
+ baseComplete,
+ WriteType.BATCH,
+ cleanup));
}
}
- if (!wrappers.isEmpty())
- {
- // Apply to local batchlog memtable in this thread
- BatchlogManager.store(Batch.createLocal(batchUUID, FBUtilities.timestampMicros(), Lists.transform(wrappers, w -> w.mutation)),
- writeCommitLog);
+ // Apply to local batchlog memtable in this thread
+ if (!nonLocalMutations.isEmpty())
+ BatchlogManager.store(Batch.createLocal(batchUUID, FBUtilities.timestampMicros(), nonLocalMutations), writeCommitLog);
- // now actually perform the writes and wait for them to complete
+ // Perform remote writes
+ if (!wrappers.isEmpty())
asyncWriteBatchedMutations(wrappers, localDataCenter, Stage.VIEW_MUTATION);
- }
-
- if (!nonPairedMutations.isEmpty())
- {
- BatchlogManager.store(Batch.createLocal(batchUUID, FBUtilities.timestampMicros(), nonPairedMutations),
- writeCommitLog);
- }
}
}
finally
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
[4/6] cassandra git commit: Merge branch 'cassandra-3.0' into
cassandra-3.11
Posted by pa...@apache.org.
Merge branch 'cassandra-3.0' into cassandra-3.11
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/eb027a1d
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/eb027a1d
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/eb027a1d
Branch: refs/heads/trunk
Commit: eb027a1de6d97f040ef9da1552e5811539e633a0
Parents: b64a4e4 1210365
Author: Paulo Motta <pa...@apache.org>
Authored: Tue Sep 12 08:31:42 2017 -0500
Committer: Paulo Motta <pa...@apache.org>
Committed: Tue Sep 12 08:33:00 2017 -0500
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../cassandra/batchlog/BatchlogManager.java | 3 +-
.../service/BatchlogResponseHandler.java | 4 +-
.../apache/cassandra/service/StorageProxy.java | 86 +++++++++-----------
4 files changed, 44 insertions(+), 50 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/eb027a1d/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index 52775e7,26b1794..099a869
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -1,16 -1,6 +1,17 @@@
-3.0.15
+3.11.1
+ * Add a compaction option to TWCS to ignore sstables overlapping checks (CASSANDRA-13418)
+ * BTree.Builder memory leak (CASSANDRA-13754)
+ * Revert CASSANDRA-10368 of supporting non-pk column filtering due to correctness (CASSANDRA-13798)
+ * Fix cassandra-stress hang issues when an error during cluster connection happens (CASSANDRA-12938)
+ * Better bootstrap failure message when blocked by (potential) range movement (CASSANDRA-13744)
+ * "ignore" option is ignored in sstableloader (CASSANDRA-13721)
+ * Deadlock in AbstractCommitLogSegmentManager (CASSANDRA-13652)
+ * Duplicate the buffer before passing it to analyser in SASI operation (CASSANDRA-13512)
+ * Properly evict pstmts from prepared statements cache (CASSANDRA-13641)
+Merged from 3.0:
+ * Fix pending view mutations handling and cleanup batchlog when there are local and remote paired mutations (CASSANDRA-13069)
* Range deletes in a CAS batch are ignored (CASSANDRA-13655)
+ * Avoid assertion error when IndexSummary > 2G (CASSANDRA-12014)
* Change repair midpoint logging for tiny ranges (CASSANDRA-13603)
* Better handle corrupt final commitlog segment (CASSANDRA-11995)
* StreamingHistogram is not thread safe (CASSANDRA-13756)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/eb027a1d/src/java/org/apache/cassandra/batchlog/BatchlogManager.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/eb027a1d/src/java/org/apache/cassandra/service/BatchlogResponseHandler.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/eb027a1d/src/java/org/apache/cassandra/service/StorageProxy.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/service/StorageProxy.java
index b8f87d9,1ce1bc5..913c3c2
--- a/src/java/org/apache/cassandra/service/StorageProxy.java
+++ b/src/java/org/apache/cassandra/service/StorageProxy.java
@@@ -819,7 -756,31 +792,34 @@@ public class StorageProxy implements St
"but this node hasn't updated its ring metadata yet. Adding mutation to " +
"local batchlog to be replayed later.",
mutation.key());
- nonPairedMutations.add(mutation);
+ continue;
+ }
+
+ // When local node is the paired endpoint just apply the mutation locally.
+ if (pairedEndpoint.get().equals(FBUtilities.getBroadcastAddress()) && StorageService.instance.isJoined())
++ {
+ try
+ {
+ mutation.apply(writeCommitLog);
+ nonLocalMutations.remove(mutation);
+ cleanup.ackMutation();
+ }
+ catch (Exception exc)
+ {
+ logger.error("Error applying local view update to keyspace {}: {}", mutation.getKeyspaceName(), mutation);
+ throw exc;
+ }
++ }
+ else
+ {
+ wrappers.add(wrapViewBatchResponseHandler(mutation,
+ consistencyLevel,
+ consistencyLevel,
+ Collections.singletonList(pairedEndpoint.get()),
+ baseComplete,
+ WriteType.BATCH,
- cleanup));
++ cleanup,
++ queryStartNanoTime));
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
[2/6] cassandra git commit: Fix pending view mutations handling and
cleanup batchlog when there are local and remote paired mutations
Posted by pa...@apache.org.
Fix pending view mutations handling and cleanup batchlog when there are local and remote paired mutations
Patch by Paulo Motta; Reviewed by Sylvain Lebresne for CASSANDRA-13069
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/12103653
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/12103653
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/12103653
Branch: refs/heads/cassandra-3.11
Commit: 12103653f313d6f1ef030a535986123ddcffea9c
Parents: e86bef4
Author: Paulo Motta <pa...@gmail.com>
Authored: Wed Dec 21 20:19:21 2016 -0200
Committer: Paulo Motta <pa...@apache.org>
Committed: Tue Sep 12 08:30:24 2017 -0500
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../cassandra/batchlog/BatchlogManager.java | 3 +-
.../service/BatchlogResponseHandler.java | 4 +-
.../apache/cassandra/service/StorageProxy.java | 82 +++++++++-----------
4 files changed, 41 insertions(+), 49 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/12103653/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 76d155e..26b1794 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
3.0.15
+ * Fix pending view mutations handling and cleanup batchlog when there are local and remote paired mutations (CASSANDRA-13069)
* Range deletes in a CAS batch are ignored (CASSANDRA-13655)
* Change repair midpoint logging for tiny ranges (CASSANDRA-13603)
* Better handle corrupt final commitlog segment (CASSANDRA-11995)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/12103653/src/java/org/apache/cassandra/batchlog/BatchlogManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/batchlog/BatchlogManager.java b/src/java/org/apache/cassandra/batchlog/BatchlogManager.java
index b614fc5..a0b614f 100644
--- a/src/java/org/apache/cassandra/batchlog/BatchlogManager.java
+++ b/src/java/org/apache/cassandra/batchlog/BatchlogManager.java
@@ -67,6 +67,7 @@ public class BatchlogManager implements BatchlogManagerMBean
private static final Logger logger = LoggerFactory.getLogger(BatchlogManager.class);
public static final BatchlogManager instance = new BatchlogManager();
+ public static final long BATCHLOG_REPLAY_TIMEOUT = Long.getLong("cassandra.batchlog.replay_timeout_in_ms", DatabaseDescriptor.getWriteRpcTimeout() * 2);
private volatile long totalBatchesReplayed = 0; // no concurrency protection necessary as only written by replay thread.
private volatile UUID lastReplayedUuid = UUIDGen.minTimeUUID(0);
@@ -284,7 +285,7 @@ public class BatchlogManager implements BatchlogManagerMBean
public static long getBatchlogTimeout()
{
- return DatabaseDescriptor.getWriteRpcTimeout() * 2; // enough time for the actual write + BM removal mutation
+ return BATCHLOG_REPLAY_TIMEOUT; // enough time for the actual write + BM removal mutation
}
private static class ReplayingBatch
http://git-wip-us.apache.org/repos/asf/cassandra/blob/12103653/src/java/org/apache/cassandra/service/BatchlogResponseHandler.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/BatchlogResponseHandler.java b/src/java/org/apache/cassandra/service/BatchlogResponseHandler.java
index ac44923..a1477e6 100644
--- a/src/java/org/apache/cassandra/service/BatchlogResponseHandler.java
+++ b/src/java/org/apache/cassandra/service/BatchlogResponseHandler.java
@@ -50,7 +50,7 @@ public class BatchlogResponseHandler<T> extends AbstractWriteResponseHandler<T>
{
wrapped.response(msg);
if (requiredBeforeFinishUpdater.decrementAndGet(this) == 0)
- cleanup.run();
+ cleanup.ackMutation();
}
public boolean isLatencyForSnitch()
@@ -107,7 +107,7 @@ public class BatchlogResponseHandler<T> extends AbstractWriteResponseHandler<T>
this.callback = callback;
}
- public void run()
+ public void ackMutation()
{
if (mutationsWaitingForUpdater.decrementAndGet(this) == 0)
callback.invoke();
http://git-wip-us.apache.org/repos/asf/cassandra/blob/12103653/src/java/org/apache/cassandra/service/StorageProxy.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java
index 6610cf7..1ce1bc5 100644
--- a/src/java/org/apache/cassandra/service/StorageProxy.java
+++ b/src/java/org/apache/cassandra/service/StorageProxy.java
@@ -729,15 +729,16 @@ public class StorageProxy implements StorageProxyMBean
else
{
List<WriteResponseHandlerWrapper> wrappers = new ArrayList<>(mutations.size());
- List<Mutation> nonPairedMutations = new LinkedList<>();
+ //non-local mutations rely on the base mutation commit-log entry for eventual consistency
+ Set<Mutation> nonLocalMutations = new HashSet<>(mutations);
Token baseToken = StorageService.instance.getTokenMetadata().partitioner.getToken(dataKey);
ConsistencyLevel consistencyLevel = ConsistencyLevel.ONE;
//Since the base -> view replication is 1:1 we only need to store the BL locally
final Collection<InetAddress> batchlogEndpoints = Collections.singleton(FBUtilities.getBroadcastAddress());
- BatchlogResponseHandler.BatchlogCleanup cleanup = new BatchlogResponseHandler.BatchlogCleanup(mutations.size(),
- () -> asyncRemoveFromBatchlog(batchlogEndpoints, batchUUID));
+ BatchlogResponseHandler.BatchlogCleanup cleanup = new BatchlogResponseHandler.BatchlogCleanup(mutations.size(), () -> asyncRemoveFromBatchlog(batchlogEndpoints, batchUUID));
+
// add a handler for each mutation - includes checking availability, but doesn't initiate any writes, yet
for (Mutation mutation : mutations)
{
@@ -746,61 +747,50 @@ public class StorageProxy implements StorageProxyMBean
Optional<InetAddress> pairedEndpoint = ViewUtils.getViewNaturalEndpoint(keyspaceName, baseToken, tk);
Collection<InetAddress> pendingEndpoints = StorageService.instance.getTokenMetadata().pendingEndpointsFor(tk, keyspaceName);
- if (pairedEndpoint.isPresent())
- {
- // When local node is the endpoint and there are no pending nodes we can
- // Just apply the mutation locally.
- if (pairedEndpoint.get().equals(FBUtilities.getBroadcastAddress())
- && pendingEndpoints.isEmpty() && StorageService.instance.isJoined())
- try
- {
- mutation.apply(writeCommitLog);
- }
- catch (Exception exc)
- {
- logger.error("Error applying local view update to keyspace {}: {}", mutation.getKeyspaceName(), mutation);
- throw exc;
- }
- else
- {
- wrappers.add(wrapViewBatchResponseHandler(mutation,
- consistencyLevel,
- consistencyLevel,
- Collections.singletonList(pairedEndpoint.get()),
- baseComplete,
- WriteType.BATCH,
- cleanup));
- }
- }
- else
+ // if there are no paired endpoints there are probably range movements going on, so we write to the local batchlog to replay later
+ if (!pairedEndpoint.isPresent())
{
- //if there are no paired endpoints there are probably range movements going on,
- //so we write to the local batchlog to replay later
if (pendingEndpoints.isEmpty())
logger.warn("Received base materialized view mutation for key {} that does not belong " +
"to this node. There is probably a range movement happening (move or decommission)," +
"but this node hasn't updated its ring metadata yet. Adding mutation to " +
"local batchlog to be replayed later.",
mutation.key());
- nonPairedMutations.add(mutation);
+ continue;
+ }
+
+ // When local node is the paired endpoint just apply the mutation locally.
+ if (pairedEndpoint.get().equals(FBUtilities.getBroadcastAddress()) && StorageService.instance.isJoined())
+ try
+ {
+ mutation.apply(writeCommitLog);
+ nonLocalMutations.remove(mutation);
+ cleanup.ackMutation();
+ }
+ catch (Exception exc)
+ {
+ logger.error("Error applying local view update to keyspace {}: {}", mutation.getKeyspaceName(), mutation);
+ throw exc;
+ }
+ else
+ {
+ wrappers.add(wrapViewBatchResponseHandler(mutation,
+ consistencyLevel,
+ consistencyLevel,
+ Collections.singletonList(pairedEndpoint.get()),
+ baseComplete,
+ WriteType.BATCH,
+ cleanup));
}
}
- if (!wrappers.isEmpty())
- {
- // Apply to local batchlog memtable in this thread
- BatchlogManager.store(Batch.createLocal(batchUUID, FBUtilities.timestampMicros(), Lists.transform(wrappers, w -> w.mutation)),
- writeCommitLog);
+ // Apply to local batchlog memtable in this thread
+ if (!nonLocalMutations.isEmpty())
+ BatchlogManager.store(Batch.createLocal(batchUUID, FBUtilities.timestampMicros(), nonLocalMutations), writeCommitLog);
- // now actually perform the writes and wait for them to complete
+ // Perform remote writes
+ if (!wrappers.isEmpty())
asyncWriteBatchedMutations(wrappers, localDataCenter, Stage.VIEW_MUTATION);
- }
-
- if (!nonPairedMutations.isEmpty())
- {
- BatchlogManager.store(Batch.createLocal(batchUUID, FBUtilities.timestampMicros(), nonPairedMutations),
- writeCommitLog);
- }
}
}
finally
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org