You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by bl...@apache.org on 2015/07/15 16:25:31 UTC
cassandra git commit: Add missing tests for CASSANDRA-9460
Repository: cassandra
Updated Branches:
refs/heads/cassandra-2.1 b16d8c9ca -> 07103ddb7
Add missing tests for CASSANDRA-9460
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/07103ddb
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/07103ddb
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/07103ddb
Branch: refs/heads/cassandra-2.1
Commit: 07103ddb7211efce8150fc91d2449cc0adbcef11
Parents: b16d8c9
Author: blerer <be...@datastax.com>
Authored: Wed Jul 15 16:22:56 2015 +0200
Committer: blerer <be...@datastax.com>
Committed: Wed Jul 15 16:22:56 2015 +0200
----------------------------------------------------------------------
.../cassandra/service/ResponseResolverTest.java | 306 +++++++++++++++++++
1 file changed, 306 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/07103ddb/test/unit/org/apache/cassandra/service/ResponseResolverTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/service/ResponseResolverTest.java b/test/unit/org/apache/cassandra/service/ResponseResolverTest.java
new file mode 100644
index 0000000..54e584d
--- /dev/null
+++ b/test/unit/org/apache/cassandra/service/ResponseResolverTest.java
@@ -0,0 +1,306 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.service;
+
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.nio.ByteBuffer;
+import java.util.*;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import org.junit.Test;
+
+import org.apache.cassandra.SchemaLoader;
+import org.apache.cassandra.db.*;
+import org.apache.cassandra.db.filter.ColumnSlice;
+import org.apache.cassandra.db.filter.SliceQueryFilter;
+import org.apache.cassandra.net.MessageIn;
+import org.apache.cassandra.net.MessagingService;
+
+import static org.apache.cassandra.Util.column;
+import static org.apache.cassandra.utils.ByteBufferUtil.bytes;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.fail;
+
+public class ResponseResolverTest extends SchemaLoader
+{
+ private final String KEYSPACE = "Keyspace1";
+ private final String TABLE = "Standard1";
+
+ @Test
+ public void testSingleMessage_RowDigestResolver() throws DigestMismatchException, UnknownHostException
+ {
+ ByteBuffer key = bytes("key");
+ ColumnFamily cf = ArrayBackedSortedColumns.factory.create(KEYSPACE, TABLE);
+ cf.addColumn(column("c1", "v1", 0));
+ Row row = new Row(key, cf);
+
+ testReadResponses(new RowDigestResolver(KEYSPACE, key), row, makeReadResponse("127.0.0.1", row));
+ }
+
+ @Test
+ public void testMultipleMessages_RowDigestResolver() throws DigestMismatchException, UnknownHostException
+ {
+ ByteBuffer key = bytes("key");
+ ColumnFamily cf = ArrayBackedSortedColumns.factory.create(KEYSPACE, TABLE);
+ cf.addColumn(column("c1", "v1", 0));
+ Row row = new Row(key, cf);
+
+ testReadResponses(new RowDigestResolver(KEYSPACE, key),
+ row,
+ makeReadResponse("127.0.0.1", row),
+ makeReadResponse("127.0.0.2", row),
+ makeReadResponse("127.0.0.3", row));
+ }
+
+ @Test(expected = DigestMismatchException.class)
+ public void testDigestMismatch() throws DigestMismatchException, UnknownHostException
+ {
+ ByteBuffer key = bytes("key");
+ ColumnFamily cf1 = ArrayBackedSortedColumns.factory.create(KEYSPACE, TABLE);
+ cf1.addColumn(column("c1", "v1", 0));
+ Row row1 = new Row(key, cf1);
+
+ ColumnFamily cf2 = ArrayBackedSortedColumns.factory.create(KEYSPACE, TABLE);
+ cf2.addColumn(column("c1", "v2", 1));
+ Row row2 = new Row(key, cf2);
+
+ testReadResponses(new RowDigestResolver(KEYSPACE, key),
+ row1,
+ makeReadResponse("127.0.0.1", row1),
+ makeReadResponse("127.0.0.2", row2),
+ makeReadResponse("127.0.0.3", row1));
+ }
+
+ @Test
+ public void testMultipleThreads_RowDigestResolver() throws DigestMismatchException, UnknownHostException, InterruptedException
+ {
+ ByteBuffer key = bytes("key");
+ ColumnFamily cf = ArrayBackedSortedColumns.factory.create(KEYSPACE, TABLE);
+ cf.addColumn(column("c1", "v1", 0));
+ Row row = new Row(key, cf);
+
+ testReadResponsesMT(new RowDigestResolver(KEYSPACE, key),
+ row,
+ makeReadResponse("127.0.0.1", row),
+ makeReadResponse("127.0.0.2", row),
+ makeReadResponse("127.0.0.3", row));
+ }
+
+ @Test
+ public void testSingleMessage_RowDataResolver() throws DigestMismatchException, UnknownHostException
+ {
+ ByteBuffer key = bytes("key");
+ ColumnFamily cf = ArrayBackedSortedColumns.factory.create(KEYSPACE, TABLE);
+ cf.addColumn(column("c1", "v1", 0));
+ Row row = new Row(key, cf);
+
+ testReadResponses(new RowDataResolver(KEYSPACE,
+ key,
+ new SliceQueryFilter(ColumnSlice.ALL_COLUMNS_ARRAY, false, 10),
+ System.currentTimeMillis()),
+ row,
+ makeReadResponse("127.0.0.1", row));
+ }
+
+ @Test
+ public void testMultipleMessages_RowDataResolver() throws DigestMismatchException, UnknownHostException
+ {
+ ByteBuffer key = bytes("key");
+ ColumnFamily cf = ArrayBackedSortedColumns.factory.create(KEYSPACE, TABLE);
+ cf.addColumn(column("c1", "v1", 0));
+ Row row = new Row(key, cf);
+
+ testReadResponses(new RowDataResolver(KEYSPACE,
+ key,
+ new SliceQueryFilter(ColumnSlice.ALL_COLUMNS_ARRAY, false, 10),
+ System.currentTimeMillis()),
+ row,
+ makeReadResponse("127.0.0.1", row),
+ makeReadResponse("127.0.0.2", row),
+ makeReadResponse("127.0.0.3", row));
+ }
+
+ @Test
+ public void testMultipleThreads_RowDataResolver() throws DigestMismatchException, UnknownHostException, InterruptedException
+ {
+ ByteBuffer key = bytes("key");
+ ColumnFamily cf = ArrayBackedSortedColumns.factory.create(KEYSPACE, TABLE);
+ cf.addColumn(column("c1", "v1", 0));
+ Row row = new Row(key, cf);
+
+ testReadResponsesMT(new RowDataResolver(KEYSPACE,
+ key,
+ new SliceQueryFilter(ColumnSlice.ALL_COLUMNS_ARRAY, false, 10),
+ System.currentTimeMillis()),
+ row,
+ makeReadResponse("127.0.0.1", row),
+ makeReadResponse("127.0.0.2", row),
+ makeReadResponse("127.0.0.3", row));
+ }
+
+ @Test
+ public void testSingleMessage_RangeSliceResolver() throws DigestMismatchException, UnknownHostException
+ {
+ ByteBuffer key = bytes("key");
+ ColumnFamily cf = ArrayBackedSortedColumns.factory.create(KEYSPACE, TABLE);
+ cf.addColumn(column("c1", "v1", 0));
+
+ Row[] expected = new Row[2];
+ for (int i = 0; i < expected.length; i++)
+ expected[i] = new Row(key, cf);
+
+ MessageIn<RangeSliceReply> message = makeRangeSlice("127.0.0.1", expected);
+
+ RangeSliceResponseResolver resolver = new RangeSliceResponseResolver(KEYSPACE, System.currentTimeMillis());
+ resolver.setSources(Collections.singletonList(message.from));
+
+ testRangeSlices(resolver, expected, message);
+ }
+
+ @Test
+ public void testMultipleMessages_RangeSliceResolver() throws DigestMismatchException, UnknownHostException
+ {
+ ByteBuffer key = bytes("key");
+ ColumnFamily cf = ArrayBackedSortedColumns.factory.create(KEYSPACE, TABLE);
+ cf.addColumn(column("c1", "v1", 0));
+
+ Row[] expected = new Row[2];
+ for (int i = 0; i < expected.length; i++)
+ expected[i] = new Row(key, cf);
+
+ List<InetAddress> sources = new ArrayList<>(3);
+ sources.add(InetAddress.getByName("127.0.0.1"));
+ sources.add(InetAddress.getByName("127.0.0.2"));
+ sources.add(InetAddress.getByName("127.0.0.3"));
+
+ RangeSliceResponseResolver resolver = new RangeSliceResponseResolver(KEYSPACE, System.currentTimeMillis());
+ resolver.setSources(sources);
+
+ testRangeSlices(resolver,
+ expected,
+ makeRangeSlice("127.0.0.1", expected),
+ makeRangeSlice("127.0.0.2", expected),
+ makeRangeSlice("127.0.0.3", expected));
+ }
+
+ private void testReadResponses(AbstractRowResolver resolver, Row expected, MessageIn<ReadResponse> ... messages) throws DigestMismatchException
+ {
+ for (MessageIn<ReadResponse> message : messages)
+ {
+ resolver.preprocess(message);
+
+ Row row = resolver.getData();
+ checkSame(expected, row);
+
+ row = resolver.resolve();
+ checkSame(expected, row);
+ }
+ }
+
+ private void testReadResponsesMT(final AbstractRowResolver resolver,
+ final Row expected,
+ final MessageIn<ReadResponse> ... messages) throws InterruptedException
+ {
+ for (MessageIn<ReadResponse> message : messages)
+ resolver.preprocess(message);
+
+ final int threadCount = 45;
+ ExecutorService executorService = Executors.newFixedThreadPool(threadCount);
+ final CountDownLatch finished = new CountDownLatch(threadCount);
+
+ for (int i = 0; i < threadCount; i++)
+ {
+ executorService.submit(new Runnable()
+ {
+ public void run()
+ {
+ try
+ {
+ Row row = resolver.getData();
+ checkSame(expected, row);
+
+ row = resolver.resolve();
+ checkSame(expected, row);
+ }
+ catch (DigestMismatchException ex)
+ {
+ fail(ex.getMessage());
+ }
+ finally
+ {
+ finished.countDown();
+ }
+ }
+ });
+ }
+
+ finished.await();
+ assertEquals(0, executorService.shutdownNow().size());
+
+ }
+
+ private void testRangeSlices(RangeSliceResponseResolver resolver, Row[] expected, MessageIn<RangeSliceReply> ... messages)
+ {
+ for (MessageIn<RangeSliceReply> message : messages)
+ {
+ resolver.preprocess(message);
+
+ List<Row> rows = resolver.getData();
+ assertNotNull(rows);
+
+ for (int i = 0; i < expected.length; i++)
+ checkSame(expected[i], rows.get(i));
+
+ Iterator<Row> rowIt = resolver.resolve().iterator();
+ assertNotNull(rowIt);
+
+ for (Row r : expected)
+ checkSame(r, rowIt.next());
+ }
+ }
+
+ private MessageIn<ReadResponse> makeReadResponse(String address, Row row) throws UnknownHostException
+ {
+ return MessageIn.create(InetAddress.getByName(address),
+ new ReadResponse(row),
+ Collections.<String, byte[]>emptyMap(),
+ MessagingService.Verb.INTERNAL_RESPONSE,
+ MessagingService.current_version);
+ }
+
+ private MessageIn<RangeSliceReply> makeRangeSlice(String address, Row ... rows) throws UnknownHostException
+ {
+ return MessageIn.create(InetAddress.getByName(address),
+ new RangeSliceReply(Arrays.asList(rows)),
+ Collections.<String, byte[]>emptyMap(),
+ MessagingService.Verb.INTERNAL_RESPONSE,
+ MessagingService.current_version);
+ }
+
+ private void checkSame(Row r1, Row r2)
+ {
+ assertEquals(r1.key, r2.key);
+ assertEquals(r1.cf, r2.cf);
+ }
+}