Repository: cassandra
Updated Branches:
  refs/heads/trunk 5250d7ffd -> 6c14c3af7


Fix NullPointerException creating digest

patch by Stefania Alborghetti; review by Benjamin Lerer for CASSANDRA-9460


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/b16d8c9c
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/b16d8c9c
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/b16d8c9c

Branch: refs/heads/trunk
Commit: b16d8c9caa8afa0c03cd6e4077d272b0c575404b
Parents: 9d44186
Author: Stefania Alborghetti <stefania.alborghe...@datastax.com>
Authored: Wed Jul 15 11:55:26 2015 +0200
Committer: blerer <benjamin.le...@datastax.com>
Committed: Wed Jul 15 11:59:23 2015 +0200

----------------------------------------------------------------------
 .../org/apache/cassandra/db/ReadResponse.java   | 26 ++++++++++++++++----
 .../service/RangeSliceResponseResolver.java     | 14 +++++------
 .../apache/cassandra/service/ReadCallback.java  |  1 +
 .../cassandra/service/RowDigestResolver.java    | 10 +++++++-
 4 files changed, 38 insertions(+), 13 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/b16d8c9c/src/java/org/apache/cassandra/db/ReadResponse.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/ReadResponse.java 
b/src/java/org/apache/cassandra/db/ReadResponse.java
index 39022a4..6a8aa8c 100644
--- a/src/java/org/apache/cassandra/db/ReadResponse.java
+++ b/src/java/org/apache/cassandra/db/ReadResponse.java
@@ -19,6 +19,7 @@ package org.apache.cassandra.db;
 
 import java.io.*;
 import java.nio.ByteBuffer;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
 
 import org.apache.cassandra.io.IVersionedSerializer;
 import org.apache.cassandra.io.util.DataOutputPlus;
@@ -32,22 +33,27 @@ import org.apache.cassandra.utils.ByteBufferUtil;
 public class ReadResponse
 {
     public static final IVersionedSerializer<ReadResponse> serializer = new 
ReadResponseSerializer();
+    private static final AtomicReferenceFieldUpdater<ReadResponse, ByteBuffer> 
digestUpdater = AtomicReferenceFieldUpdater.newUpdater(ReadResponse.class, 
ByteBuffer.class, "digest");
 
     private final Row row;
-    private final ByteBuffer digest;
+    private volatile ByteBuffer digest;
 
     public ReadResponse(ByteBuffer digest)
     {
+        this(null, digest);
         assert digest != null;
-        this.digest= digest;
-        this.row = null;
     }
 
     public ReadResponse(Row row)
     {
+        this(row, null);
         assert row != null;
+    }
+
+    public ReadResponse(Row row, ByteBuffer digest)
+    {
         this.row = row;
-        this.digest = null;
+        this.digest = digest;
     }
 
     public Row row()
@@ -60,9 +66,19 @@ public class ReadResponse
         return digest;
     }
 
+    public void setDigest(ByteBuffer digest)
+    {
+        ByteBuffer curr = this.digest;
+        if (!digestUpdater.compareAndSet(this, curr, digest))
+        {
+            assert digest.equals(this.digest) :
+                String.format("Digest mismatch : %s vs %s", digest.array(), 
this.digest.array());
+        }
+    }
+
     public boolean isDigestQuery()
     {
-        return digest != null;
+        return digest != null && row == null;
     }
 }
 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/b16d8c9c/src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java
----------------------------------------------------------------------
diff --git 
a/src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java 
b/src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java
index 640681b..4242481 100644
--- a/src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java
+++ b/src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java
@@ -50,8 +50,8 @@ public class RangeSliceResponseResolver implements 
IResponseResolver<RangeSliceR
     private final String keyspaceName;
     private final long timestamp;
     private List<InetAddress> sources;
-    protected final Collection<MessageIn<RangeSliceReply>> responses = new 
ConcurrentLinkedQueue<MessageIn<RangeSliceReply>>();
-    public final List<AsyncOneResponse> repairResults = new 
ArrayList<AsyncOneResponse>();
+    protected final Queue<MessageIn<RangeSliceReply>> responses = new 
ConcurrentLinkedQueue<>();
+    public final List<AsyncOneResponse> repairResults = new ArrayList<>();
 
     public RangeSliceResponseResolver(String keyspaceName, long timestamp)
     {
@@ -66,15 +66,15 @@ public class RangeSliceResponseResolver implements 
IResponseResolver<RangeSliceR
 
     public List<Row> getData()
     {
-        MessageIn<RangeSliceReply> response = responses.iterator().next();
-        return response.payload.rows;
+        assert !responses.isEmpty();
+        return responses.peek().payload.rows;
     }
 
     // Note: this would deserialize the response a 2nd time if getData was 
called first.
     // (this is not currently an issue since we don't do read repair for range 
queries.)
     public Iterable<Row> resolve()
     {
-        ArrayList<RowIterator> iters = new 
ArrayList<RowIterator>(responses.size());
+        ArrayList<RowIterator> iters = new ArrayList<>(responses.size());
         int n = 0;
         for (MessageIn<RangeSliceReply> response : responses)
         {
@@ -86,7 +86,7 @@ public class RangeSliceResponseResolver implements 
IResponseResolver<RangeSliceR
         // TODO do we need to call close?
         CloseableIterator<Row> iter = MergeIterator.get(iters, pairComparator, 
new Reducer());
 
-        List<Row> resolvedRows = new ArrayList<Row>(n);
+        List<Row> resolvedRows = new ArrayList<>(n);
         while (iter.hasNext())
             resolvedRows.add(iter.next());
 
@@ -129,7 +129,7 @@ public class RangeSliceResponseResolver implements 
IResponseResolver<RangeSliceR
 
     private class Reducer extends MergeIterator.Reducer<Pair<Row,InetAddress>, 
Row>
     {
-        List<ColumnFamily> versions = new 
ArrayList<ColumnFamily>(sources.size());
+        List<ColumnFamily> versions = new ArrayList<>(sources.size());
         List<InetAddress> versionSources = new 
ArrayList<InetAddress>(sources.size());
         DecoratedKey key;
 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/b16d8c9c/src/java/org/apache/cassandra/service/ReadCallback.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/ReadCallback.java 
b/src/java/org/apache/cassandra/service/ReadCallback.java
index cf9be55..e0646a9 100644
--- a/src/java/org/apache/cassandra/service/ReadCallback.java
+++ b/src/java/org/apache/cassandra/service/ReadCallback.java
@@ -121,6 +121,7 @@ public class ReadCallback<TMessage, TResolved> implements 
IAsyncCallback<TMessag
         if (n >= blockfor && resolver.isDataPresent())
         {
             condition.signalAll();
+
             // kick off a background digest comparison if this is a result 
that (may have) arrived after
             // the original resolve that get() kicks off as soon as the 
condition is signaled
             if (blockfor < endpoints.size() && n == endpoints.size())

http://git-wip-us.apache.org/repos/asf/cassandra/blob/b16d8c9c/src/java/org/apache/cassandra/service/RowDigestResolver.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/RowDigestResolver.java 
b/src/java/org/apache/cassandra/service/RowDigestResolver.java
index 21b16bf..7f2e17d 100644
--- a/src/java/org/apache/cassandra/service/RowDigestResolver.java
+++ b/src/java/org/apache/cassandra/service/RowDigestResolver.java
@@ -41,7 +41,12 @@ public class RowDigestResolver extends AbstractRowResolver
         {
             ReadResponse result = message.payload;
             if (!result.isDigestQuery())
+            {
+                if (result.digest() == null)
+                    result.setDigest(ColumnFamily.digest(result.row().cf));
+
                 return result.row();
+            }
         }
         return null;
     }
@@ -81,7 +86,10 @@ public class RowDigestResolver extends AbstractRowResolver
             {
                 // note that this allows for multiple data replies, 
post-CASSANDRA-5932
                 data = response.row().cf;
-                newDigest = ColumnFamily.digest(data);
+                if (response.digest() == null)
+                    message.payload.setDigest(ColumnFamily.digest(data));
+
+                newDigest = response.digest();
             }
 
             if (digest == null)

Reply via email to