Repository: cassandra Updated Branches: refs/heads/trunk 4047dd121 -> 36958f3ca
Fixes ReadMessageTest.testNoCommitLog for test-compression Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/36958f3c Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/36958f3c Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/36958f3c Branch: refs/heads/trunk Commit: 36958f3cab74c9b7baca94b5e7d31de0cbd42d40 Parents: 4047dd1 Author: Branimir Lambov <[email protected]> Authored: Thu May 7 17:27:18 2015 -0700 Committer: Jonathan Ellis <[email protected]> Committed: Thu May 7 17:27:18 2015 -0700 ---------------------------------------------------------------------- .../apache/cassandra/db/ReadMessageTest.java | 84 ++++---------------- .../db/commitlog/CommitLogTestReplayer.java | 79 ++++++++++++++++++ 2 files changed, 95 insertions(+), 68 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/cassandra/blob/36958f3c/test/unit/org/apache/cassandra/db/ReadMessageTest.java ---------------------------------------------------------------------- diff --git a/test/unit/org/apache/cassandra/db/ReadMessageTest.java b/test/unit/org/apache/cassandra/db/ReadMessageTest.java index d32df49..34f25a1 100644 --- a/test/unit/org/apache/cassandra/db/ReadMessageTest.java +++ b/test/unit/org/apache/cassandra/db/ReadMessageTest.java @@ -25,13 +25,14 @@ import java.nio.ByteBuffer; import java.util.SortedSet; import java.util.TreeSet; +import com.google.common.base.Predicate; import org.junit.BeforeClass; import org.junit.Test; import org.apache.cassandra.SchemaLoader; import org.apache.cassandra.Util; -import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.config.KSMetaData; +import org.apache.cassandra.db.commitlog.CommitLogTestReplayer; import org.apache.cassandra.db.composites.*; import org.apache.cassandra.db.filter.NamesQueryFilter; import org.apache.cassandra.db.filter.SliceQueryFilter; @@ -132,81 +133,28 @@ public class ReadMessageTest rm.add("Standard1", Util.cellname("commit2"), ByteBufferUtil.bytes("abcd"), 0); rm.apply(); - boolean commitLogMessageFound = false; - boolean noCommitLogMessageFound = false; - - File commitLogDir = new File(DatabaseDescriptor.getCommitLogLocation()); - - byte[] commitBytes = "commit".getBytes("UTF-8"); - - for(String filename : commitLogDir.list()) - { - BufferedInputStream is = null; - try - { - is = new BufferedInputStream(new FileInputStream(commitLogDir.getAbsolutePath()+File.separator+filename)); - - if (!isEmptyCommitLog(is)) - { - while (findPatternInStream(commitBytes, is)) - { - char c = (char)is.read(); - - if (c == '1') - commitLogMessageFound = true; - else if (c == '2') - noCommitLogMessageFound = true; - } - } - } - finally - { - if (is != null) - is.close(); - } - } + Checker checker = new Checker(); + CommitLogTestReplayer.examineCommitLog(checker); - assertTrue(commitLogMessageFound); - assertFalse(noCommitLogMessageFound); + assertTrue(checker.commitLogMessageFound); + assertFalse(checker.noCommitLogMessageFound); } - private boolean isEmptyCommitLog(BufferedInputStream is) throws IOException + static class Checker implements Predicate<Mutation> { - DataInputStream dis = new DataInputStream(is); - byte[] lookahead = new byte[100]; - - dis.mark(100); - dis.readFully(lookahead); - dis.reset(); - - for (int i = 0; i < 100; i++) - { - if (lookahead[i] != 0) - return false; - } - - return true; - } - - private boolean findPatternInStream(byte[] pattern, InputStream is) throws IOException - { - int patternOffset = 0; + boolean commitLogMessageFound = false; + boolean noCommitLogMessageFound = false; - int b = is.read(); - while (b != -1) + public boolean apply(Mutation mutation) { - if (pattern[patternOffset] == ((byte) b)) + for (ColumnFamily cf : mutation.getColumnFamilies()) { - patternOffset++; - if (patternOffset == pattern.length) - return true; + if (cf.getColumn(Util.cellname("commit1")) != null) + commitLogMessageFound = true; + if (cf.getColumn(Util.cellname("commit2")) != null) + noCommitLogMessageFound = true; } - else - patternOffset = 0; - - b = is.read(); + return true; } - - return false; } } http://git-wip-us.apache.org/repos/asf/cassandra/blob/36958f3c/test/unit/org/apache/cassandra/db/commitlog/CommitLogTestReplayer.java ---------------------------------------------------------------------- diff --git a/test/unit/org/apache/cassandra/db/commitlog/CommitLogTestReplayer.java b/test/unit/org/apache/cassandra/db/commitlog/CommitLogTestReplayer.java new file mode 100644 index 0000000..4ad49ec --- /dev/null +++ b/test/unit/org/apache/cassandra/db/commitlog/CommitLogTestReplayer.java @@ -0,0 +1,79 @@ +/* +* 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.db.commitlog; + +import java.io.DataInputStream; +import java.io.File; +import java.io.IOException; + +import com.google.common.base.Predicate; + +import org.junit.Assert; + +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.db.ColumnSerializer; +import org.apache.cassandra.db.Mutation; +import org.apache.cassandra.io.util.FastByteArrayInputStream; + +/** + * Utility class for tests needing to examine the commitlog contents. + */ +public class CommitLogTestReplayer extends CommitLogReplayer +{ + static public void examineCommitLog(Predicate<Mutation> processor) throws IOException + { + CommitLog.instance.sync(true); + + CommitLogTestReplayer replayer = new CommitLogTestReplayer(processor); + File commitLogDir = new File(DatabaseDescriptor.getCommitLogLocation()); + replayer.recover(commitLogDir.listFiles()); + } + + final private Predicate<Mutation> processor; + + public CommitLogTestReplayer(Predicate<Mutation> processor) + { + this(ReplayPosition.NONE, processor); + } + + public CommitLogTestReplayer(ReplayPosition discardedPos, Predicate<Mutation> processor) + { + super(discardedPos, null, ReplayFilter.create()); + this.processor = processor; + } + + @Override + void replayMutation(byte[] inputBuffer, int size, final long entryLocation, final CommitLogDescriptor desc) + { + FastByteArrayInputStream bufIn = new FastByteArrayInputStream(inputBuffer, 0, size); + Mutation mutation; + try + { + mutation = Mutation.serializer.deserialize(new DataInputStream(bufIn), + desc.getMessagingVersion(), + ColumnSerializer.Flag.LOCAL); + Assert.assertTrue(processor.apply(mutation)); + } + catch (IOException e) + { + // Test fails. + throw new AssertionError(e); + } + } +}
