natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Remove timing window in test case 

patch by Chris Lohfink; reviewed by Stefan Podkowinski for CASSANDRA-12875
This commit is contained in:
Chris Lohfink 2016-11-09 08:41:41 -06:00 committed by Jason Brown
parent d5dbd0c756
commit f87d074cd2
4 changed files with 14 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CHANGES.txt
View File

@ -1,4 +1,6 @@
3.10
* Remove timing window in test case (CASSANDRA-12875)
* Resolve unit testing without JCE security libraries installed (CASSANDRA-12945)
* Fix inconsistencies in cassandra-stress load balancing policy (CASSANDRA-12919)
* Fix validation of non-frozen UDT cells (CASSANDRA-12916)
* Don't shut down socket input/output on StreamSession (CASSANDRA-12903)

5
src/java/org/apache/cassandra/net/IncomingTcpConnection.java
View File

@ -39,6 +39,7 @@ import org.apache.cassandra.config.Config;
import org.xerial.snappy.SnappyInputStream;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.UnknownColumnFamilyException;
import org.apache.cassandra.db.monitoring.ApproximateTime;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataInputPlus.DataInputStreamPlus;
import org.apache.cassandra.io.util.NIODataInputStream;
@ -187,8 +188,8 @@ public class IncomingTcpConnection extends FastThreadLocalThread implements Clos
id = Integer.parseInt(input.readUTF());
else
id = input.readInt();
MessageIn message = MessageIn.read(input, version, id, MessageIn.readConstructionTime(from, input));
long currentTime = ApproximateTime.currentTimeMillis();
MessageIn message = MessageIn.read(input, version, id, MessageIn.readConstructionTime(from, input, currentTime));
if (message == null)
{
// callback expired; nothing to do

4
src/java/org/apache/cassandra/net/MessageIn.java
View File

@ -124,10 +124,8 @@ public class MessageIn<T>
return MessageIn.create(from, payload, parameters, verb, version, constructionTime);
}
public static long readConstructionTime(InetAddress from, DataInputPlus input) throws IOException
public static long readConstructionTime(InetAddress from, DataInputPlus input, long currentTime) throws IOException
{
long currentTime = ApproximateTime.currentTimeMillis();
// Reconstruct the message construction time sent by the remote host (we sent only the lower 4 bytes, assuming the
// higher 4 bytes wouldn't change between the sender and receiver)
int partial = input.readInt(); // make sure to readInt, even if cross_node_to is not enabled

20
test/unit/org/apache/cassandra/net/MessagingServiceTest.java
View File

@ -34,20 +34,18 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import com.google.common.collect.Iterables;
import com.codahale.metrics.Timer;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.monitoring.ApproximateTime;
import org.apache.cassandra.io.util.DataInputPlus.DataInputStreamPlus;
import org.apache.cassandra.io.util.DataOutputStreamPlus;
import org.apache.cassandra.io.util.WrappedDataOutputStreamPlus;
import org.caffinitas.ohc.histo.EstimatedHistogram;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.config.DatabaseDescriptor;
import static org.junit.Assert.*;
public class MessagingServiceTest
@ -100,15 +98,13 @@ public class MessagingServiceTest
public void testDCLatency() throws Exception
{
int latency = 100;
ConcurrentHashMap<String, Timer> dcLatency = MessagingService.instance().metrics.dcLatency;
dcLatency.clear();
long now = System.currentTimeMillis();
long now = ApproximateTime.currentTimeMillis();
long sentAt = now - latency;
assertNull(dcLatency.get("datacenter1"));
addDCLatency(sentAt);
addDCLatency(sentAt, now);
assertNotNull(dcLatency.get("datacenter1"));
assertEquals(1, dcLatency.get("datacenter1").getCount());
long expectedBucket = bucketOffsets[Math.abs(Arrays.binarySearch(bucketOffsets, TimeUnit.MILLISECONDS.toNanos(latency))) - 1];
@ -124,11 +120,11 @@ public class MessagingServiceTest
ConcurrentHashMap<String, Timer> dcLatency = MessagingService.instance().metrics.dcLatency;
dcLatency.clear();
long now = System.currentTimeMillis();
long now = ApproximateTime.currentTimeMillis();
long sentAt = now - latency;
assertNull(dcLatency.get("datacenter1"));
addDCLatency(sentAt);
addDCLatency(sentAt, now);
assertNull(dcLatency.get("datacenter1"));
}
@ -221,7 +217,7 @@ public class MessagingServiceTest
assertFalse(MockBackPressureStrategy.applied);
}
private static void addDCLatency(long sentAt) throws IOException
private static void addDCLatency(long sentAt, long nowTime) throws IOException
{
ByteArrayOutputStream baos = new ByteArrayOutputStream();
try (DataOutputStreamPlus out = new WrappedDataOutputStreamPlus(baos))
@ -229,7 +225,7 @@ public class MessagingServiceTest
out.writeInt((int) sentAt);
}
DataInputStreamPlus in = new DataInputStreamPlus(new ByteArrayInputStream(baos.toByteArray()));
MessageIn.readConstructionTime(InetAddress.getLocalHost(), in);
MessageIn.readConstructionTime(InetAddress.getLocalHost(), in, nowTime);
}
public static class MockBackPressureStrategy implements BackPressureStrategy<MockBackPressureStrategy.MockBackPressureState>