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

Use node's host id in place of counter ids 

patch by Aleksey Yeschenko; reviewed by Sylvain Lebresne for
CASSANDRA-7366
This commit is contained in:
Aleksey Yeschenko 2014-06-11 17:21:35 -05:00
parent 5fe7557627
commit 99594cd687
7 changed files with 9 additions and 152 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
2.1.0
* Use node's host id in place of counter ids (CASSANDRA-7366)
* Explicitly use Long.MAX_VALUE timestamp for counter deletions
(CASSANDRA-7346)
* Fix native protocol CAS batches (CASSANDRA-7337)

6
src/java/org/apache/cassandra/config/CFMetaData.java
View File

@ -144,12 +144,6 @@ public final class CFMetaData
+ "PRIMARY KEY (table_name, index_name)"
+ ") WITH COMPACT STORAGE AND COMMENT='indexes that have been completed'");
public static final CFMetaData CounterIdCf = compile("CREATE TABLE \"" + SystemKeyspace.COUNTER_ID_CF + "\" ("
+ "key text,"
+ "id timeuuid,"
+ "PRIMARY KEY (key, id)"
+ ") WITH COMPACT STORAGE AND COMMENT='counter node IDs'");
public static final CFMetaData SchemaKeyspacesCf = compile("CREATE TABLE " + SystemKeyspace.SCHEMA_KEYSPACES_CF + " ("
+ "keyspace_name text PRIMARY KEY,"
+ "durable_writes boolean,"

1
src/java/org/apache/cassandra/config/KSMetaData.java
View File

@ -96,7 +96,6 @@ public final class KSMetaData
CFMetaData.PeerEventsCf,
CFMetaData.HintsCf,
CFMetaData.IndexCf,
CFMetaData.CounterIdCf,
CFMetaData.SchemaKeyspacesCf,
CFMetaData.SchemaColumnFamiliesCf,
CFMetaData.SchemaColumnsCf,

49
src/java/org/apache/cassandra/db/SystemKeyspace.java
View File

@ -29,7 +29,6 @@ import com.google.common.collect.HashMultimap;
import com.google.common.collect.Iterables;
import com.google.common.collect.SetMultimap;
import com.google.common.collect.Sets;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -44,7 +43,6 @@ import org.apache.cassandra.db.columniterator.IdentityQueryFilter;
import org.apache.cassandra.db.compaction.CompactionHistoryTabularData;
import org.apache.cassandra.db.commitlog.ReplayPosition;
import org.apache.cassandra.db.composites.Composite;
import org.apache.cassandra.db.composites.Composites;
import org.apache.cassandra.db.filter.QueryFilter;
import org.apache.cassandra.db.marshal.*;
import org.apache.cassandra.dht.Range;
@ -73,7 +71,6 @@ public class SystemKeyspace
public static final String PEER_EVENTS_CF = "peer_events";
public static final String LOCAL_CF = "local";
public static final String INDEX_CF = "IndexInfo";
public static final String COUNTER_ID_CF = "NodeIdInfo";
public static final String HINTS_CF = "hints";
public static final String RANGE_XFERS_CF = "range_xfers";
public static final String BATCHLOG_CF = "batchlog";
@ -89,7 +86,6 @@ public class SystemKeyspace
public static final String COMPACTION_HISTORY_CF = "compaction_history";
private static final String LOCAL_KEY = "local";
private static final ByteBuffer ALL_LOCAL_NODE_ID_KEY = ByteBufferUtil.bytes("Local");
public static final List<String> allSchemaCfs = Arrays.asList(SCHEMA_KEYSPACES_CF,
SCHEMA_COLUMNFAMILIES_CF,
@ -688,19 +684,15 @@ public class SystemKeyspace
*/
public static UUID getLocalHostId()
{
UUID hostId = null;
String req = "SELECT host_id FROM system.%s WHERE key='%s'";
UntypedResultSet result = executeInternal(String.format(req, LOCAL_CF, LOCAL_KEY));
// Look up the Host UUID (return it if found)
if (!result.isEmpty() && result.one().has("host_id"))
{
return result.one().getUUID("host_id");
}
// ID not found, generate a new one, persist, and then return it.
hostId = UUID.randomUUID();
UUID hostId = UUID.randomUUID();
logger.warn("No host ID found, created {} (Note: This should happen exactly once per node).", hostId);
return setLocalHostId(hostId);
}
@ -715,45 +707,6 @@ public class SystemKeyspace
return hostId;
}
/**
* Read the current local node id from the system keyspace or null if no
* such node id is recorded.
*/
public static CounterId getCurrentLocalCounterId()
{
Keyspace keyspace = Keyspace.open(Keyspace.SYSTEM_KS);
// Get the last CounterId (since CounterId are timeuuid is thus ordered from the older to the newer one)
QueryFilter filter = QueryFilter.getSliceFilter(decorate(ALL_LOCAL_NODE_ID_KEY),
COUNTER_ID_CF,
Composites.EMPTY,
Composites.EMPTY,
true,
1,
System.currentTimeMillis());
ColumnFamily cf = keyspace.getColumnFamilyStore(COUNTER_ID_CF).getColumnFamily(filter);
if (cf != null && cf.hasColumns())
return CounterId.wrap(cf.iterator().next().name().toByteBuffer());
else
return null;
}
/**
* Write a new current local node id to the system keyspace.
*
* @param newCounterId the new current local node id to record
* @param now microsecond time stamp.
*/
public static void writeCurrentLocalCounterId(CounterId newCounterId, long now)
{
ByteBuffer ip = ByteBuffer.wrap(FBUtilities.getBroadcastAddress().getAddress());
ColumnFamily cf = ArrayBackedSortedColumns.factory.create(Keyspace.SYSTEM_KS, COUNTER_ID_CF);
cf.addColumn(new BufferCell(cf.getComparator().makeCellName(newCounterId.bytes()), ip, now));
new Mutation(Keyspace.SYSTEM_KS, ALL_LOCAL_NODE_ID_KEY, cf).apply();
forceBlockingFlush(COUNTER_ID_CF);
}
/**
* @param cfName The name of the ColumnFamily responsible for part of the schema (keyspace, ColumnFamily, columns)
* @return CFS responsible to hold low-level serialized schema

17
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -560,17 +560,6 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
}
}
if (Boolean.parseBoolean(System.getProperty("cassandra.renew_counter_id", "false")))
{
logger.info("Renewing local node id (as requested)");
CounterId.renewLocalId();
}
// Can't do this in CassandraDaemon before the SS start b/c local counter id can be renewed afterwards.
for (ColumnFamilyStore cfs : ColumnFamilyStore.all())
if (cfs.metadata.isCounter())
cfs.initCounterCache();
// daemon threads, like our executors', continue to run while shutdown hooks are invoked
Thread drainOnShutdown = new Thread(new WrappedRunnable()
{
@ -626,6 +615,12 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
Runtime.getRuntime().addShutdownHook(drainOnShutdown);
prepareToJoin();
// Has to be called after the host id has potentially changed in prepareToJoin().
for (ColumnFamilyStore cfs : ColumnFamilyStore.all())
if (cfs.metadata.isCounter())
cfs.initCounterCache();
if (Boolean.parseBoolean(System.getProperty("cassandra.join_ring", "true")))
{
joinTokenRing(delay);

38
src/java/org/apache/cassandra/utils/CounterId.java
View File

@ -20,15 +20,10 @@ package org.apache.cassandra.utils;
import java.nio.ByteBuffer;
import java.util.concurrent.atomic.AtomicReference;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.db.SystemKeyspace;
public class CounterId implements Comparable<CounterId>
{
private static final Logger logger = LoggerFactory.getLogger(CounterId.class);
public static final int LENGTH = 16; // we assume a fixed length size for all CounterIds
// Lazy holder because this opens the system keyspace and we want to avoid
@ -50,16 +45,6 @@ public class CounterId implements Comparable<CounterId>
return localId().get();
}
/**
* Renew the local counter id.
* To use only when this strictly necessary, as using this will make all
* counter context grow with time.
*/
public static synchronized void renewLocalId()
{
localId().renew(FBUtilities.timestampMicros());
}
/**
* Function for test purposes, do not use otherwise.
* Pack an int in a valid CounterId so that the resulting ids respects the
@ -150,28 +135,7 @@ public class CounterId implements Comparable<CounterId>
LocalCounterIdHolder()
{
CounterId id = SystemKeyspace.getCurrentLocalCounterId();
if (id == null)
{
// no recorded local counter id, generating a new one and saving it
id = generate();
logger.info("No saved local counter id, using newly generated: {}", id);
SystemKeyspace.writeCurrentLocalCounterId(id, FBUtilities.timestampMicros());
}
else
{
logger.info("Using saved local counter id: {}", id);
}
current = new AtomicReference<>(id);
}
synchronized void renew(long now)
{
CounterId newCounterId = generate();
SystemKeyspace.writeCurrentLocalCounterId(newCounterId, now);
current.set(newCounterId);
current = new AtomicReference<>(wrap(ByteBufferUtil.bytes(SystemKeyspace.getLocalHostId())));
}
CounterId get()

49
test/unit/org/apache/cassandra/utils/CounterIdTest.java
View File

@ -1,49 +0,0 @@
/**
* 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.utils;
import org.junit.Test;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.db.SystemKeyspace;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
public class CounterIdTest extends SchemaLoader
{
@Test
public void testGetCurrentIdFromSystemKeyspace()
{
// Renewing a bunch of times and checking we get the same thing from
// the system keyspace that what is in memory
CounterId id0 = CounterId.getLocalId();
assertEquals(id0, SystemKeyspace.getCurrentLocalCounterId());
CounterId.renewLocalId();
CounterId id1 = CounterId.getLocalId();
assertEquals(id1, SystemKeyspace.getCurrentLocalCounterId());
assertTrue(id1.compareTo(id0) == 1);
CounterId.renewLocalId();
CounterId id2 = CounterId.getLocalId();
assertEquals(id2, SystemKeyspace.getCurrentLocalCounterId());
assertTrue(id2.compareTo(id1) == 1);
}
}