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

Merge branch 'cassandra-3.0' into trunk

This commit is contained in:
Aleksey Yeschenko 2015-11-04 13:13:32 +00:00
parent b14423c8b4 1492be07cd
commit 81e1b15dbd
3 changed files with 48 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -5,6 +5,7 @@
3.0
* Guard batchlog replay against integer division by zero (CASSANDRA-9223)
* Fix bug when adding a column to thrift with the same name than a primary key (CASSANDRA-10608)
* Add client address argument to IAuthenticator::newSaslNegotiator (CASSANDRA-8068)
* Fix implementation of LegacyLayout.LegacyBoundComparator (CASSANDRA-10602)

8
src/java/org/apache/cassandra/batchlog/BatchlogManager.java
View File

@ -178,7 +178,13 @@ public class BatchlogManager implements BatchlogManagerMBean
// rate limit is in bytes per second. Uses Double.MAX_VALUE if disabled (set to 0 in cassandra.yaml).
// max rate is scaled by the number of nodes in the cluster (same as for HHOM - see CASSANDRA-5272).
int throttleInKB = DatabaseDescriptor.getBatchlogReplayThrottleInKB() / StorageService.instance.getTokenMetadata().getAllEndpoints().size();
int endpointsCount = StorageService.instance.getTokenMetadata().getAllEndpoints().size();
if (endpointsCount <= 0)
{
logger.trace("Replay cancelled as there are no peers in the ring.");
return;
}
int throttleInKB = DatabaseDescriptor.getBatchlogReplayThrottleInKB() / endpointsCount;
RateLimiter rateLimiter = RateLimiter.create(throttleInKB == 0 ? Double.MAX_VALUE : throttleInKB * 1024);
UUID limitUuid = UUIDGen.maxTimeUUID(System.currentTimeMillis() - getBatchlogTimeout());

40
test/unit/org/apache/cassandra/batchlog/BatchlogManagerTest.java
View File

@ -24,6 +24,7 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import com.google.common.collect.Lists;
import org.junit.*;
import org.apache.cassandra.SchemaLoader;
@ -457,4 +458,43 @@ public class BatchlogManagerTest
assertNotNull(result);
assertEquals(0L, result.one().getLong("count"));
}
// CASSANRDA-9223
@Test
public void testReplayWithNoPeers() throws Exception
{
StorageService.instance.getTokenMetadata().removeEndpoint(InetAddress.getByName("127.0.0.1"));
long initialAllBatches = BatchlogManager.instance.countAllBatches();
long initialReplayedBatches = BatchlogManager.instance.getTotalBatchesReplayed();
CFMetaData cfm = Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF_STANDARD1).metadata;
long timestamp = (System.currentTimeMillis() - DatabaseDescriptor.getWriteRpcTimeout() * 2) * 1000;
UUID uuid = UUIDGen.getTimeUUID();
// Add a batch with 10 mutations
List<Mutation> mutations = new ArrayList<>(10);
for (int j = 0; j < 10; j++)
{
mutations.add(new RowUpdateBuilder(cfm, FBUtilities.timestampMicros(), ByteBufferUtil.bytes(j))
.clustering("name" + j)
.add("val", "val" + j)
.build());
}
BatchlogManager.store(Batch.createLocal(uuid, timestamp, mutations));
assertEquals(1, BatchlogManager.instance.countAllBatches() - initialAllBatches);
// Flush the batchlog to disk (see CASSANDRA-6822).
Keyspace.open(SystemKeyspace.NAME).getColumnFamilyStore(SystemKeyspace.BATCHES).forceBlockingFlush();
assertEquals(1, BatchlogManager.instance.countAllBatches() - initialAllBatches);
assertEquals(0, BatchlogManager.instance.getTotalBatchesReplayed() - initialReplayedBatches);
// Force batchlog replay and wait for it to complete.
BatchlogManager.instance.startBatchlogReplay().get();
// Replay should be cancelled as there are no peers in the ring.
assertEquals(1, BatchlogManager.instance.countAllBatches() - initialAllBatches);
}
}