diff --git a/CHANGES.txt b/CHANGES.txt index 7d20ab3a97..0520477e27 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -6,6 +6,7 @@ * Duplicate the buffer before passing it to analyser in SASI operation (CASSANDRA-13512) * Properly evict pstmts from prepared statements cache (CASSANDRA-13641) Merged from 3.0: + * Gossip thread slows down when using batch commit log (CASSANDRA-12966) * Randomize batchlog endpoint selection with only 1 or 2 racks (CASSANDRA-12884) * Fix digest calculation for counter cells (CASSANDRA-13750) * Fix ColumnDefinition.cellValueType() for non-frozen collection and change SSTabledump to use type.toJSONString() (CASSANDRA-13573) diff --git a/src/java/org/apache/cassandra/db/SystemKeyspace.java b/src/java/org/apache/cassandra/db/SystemKeyspace.java index 6c453295da..ac51662707 100644 --- a/src/java/org/apache/cassandra/db/SystemKeyspace.java +++ b/src/java/org/apache/cassandra/db/SystemKeyspace.java @@ -23,11 +23,13 @@ import java.io.IOException; import java.net.InetAddress; import java.nio.ByteBuffer; import java.util.*; +import java.util.concurrent.ExecutorService; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; import java.util.stream.StreamSupport; import javax.management.openmbean.OpenDataException; import javax.management.openmbean.TabularData; +import java.util.concurrent.Future; import com.google.common.collect.HashMultimap; import com.google.common.collect.ImmutableMap; @@ -38,6 +40,10 @@ import com.google.common.io.ByteStreams; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.google.common.util.concurrent.Futures; + +import org.apache.cassandra.concurrent.Stage; +import org.apache.cassandra.concurrent.StageManager; import org.apache.cassandra.config.CFMetaData; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.config.SchemaConstants; @@ -728,29 +734,29 @@ public final class SystemKeyspace /** * Record tokens being used by another node */ - public static synchronized void updateTokens(InetAddress ep, Collection tokens) + public static Future updateTokens(final InetAddress ep, final Collection tokens, ExecutorService executorService) { if (ep.equals(FBUtilities.getBroadcastAddress())) - return; + return Futures.immediateFuture(null); String req = "INSERT INTO system.%s (peer, tokens) VALUES (?, ?)"; - executeInternal(String.format(req, PEERS), ep, tokensAsSet(tokens)); + return executorService.submit((Runnable) () -> executeInternal(String.format(req, PEERS), ep, tokensAsSet(tokens))); } - public static synchronized void updatePreferredIP(InetAddress ep, InetAddress preferred_ip) + public static void updatePreferredIP(InetAddress ep, InetAddress preferred_ip) { String req = "INSERT INTO system.%s (peer, preferred_ip) VALUES (?, ?)"; executeInternal(String.format(req, PEERS), ep, preferred_ip); forceBlockingFlush(PEERS); } - public static synchronized void updatePeerInfo(InetAddress ep, String columnName, Object value) + public static Future updatePeerInfo(final InetAddress ep, final String columnName, final Object value, ExecutorService executorService) { if (ep.equals(FBUtilities.getBroadcastAddress())) - return; + return Futures.immediateFuture(null); String req = "INSERT INTO system.%s (peer, %s) VALUES (?, ?)"; - executeInternal(String.format(req, PEERS, columnName), ep, value); + return executorService.submit((Runnable) () -> executeInternal(String.format(req, PEERS, columnName), ep, value)); } public static synchronized void updateHintsDropped(InetAddress ep, UUID timePeriod, int value) @@ -789,7 +795,7 @@ public final class SystemKeyspace /** * Remove stored tokens being used by another node */ - public static synchronized void removeEndpoint(InetAddress ep) + public static void removeEndpoint(InetAddress ep) { String req = "DELETE FROM system.%s WHERE peer = ?"; executeInternal(String.format(req, PEERS), ep); diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index cbf69b44a8..3eb3732aaf 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -1972,23 +1972,24 @@ public class StorageService extends NotificationBroadcasterSupport implements IE if (getTokenMetadata().isMember(endpoint)) { + final ExecutorService executor = StageManager.getStage(Stage.MUTATION); switch (state) { case RELEASE_VERSION: - SystemKeyspace.updatePeerInfo(endpoint, "release_version", value.value); + SystemKeyspace.updatePeerInfo(endpoint, "release_version", value.value, executor); break; case DC: updateTopology(endpoint); - SystemKeyspace.updatePeerInfo(endpoint, "data_center", value.value); + SystemKeyspace.updatePeerInfo(endpoint, "data_center", value.value, executor); break; case RACK: updateTopology(endpoint); - SystemKeyspace.updatePeerInfo(endpoint, "rack", value.value); + SystemKeyspace.updatePeerInfo(endpoint, "rack", value.value, executor); break; case RPC_ADDRESS: try { - SystemKeyspace.updatePeerInfo(endpoint, "rpc_address", InetAddress.getByName(value.value)); + SystemKeyspace.updatePeerInfo(endpoint, "rpc_address", InetAddress.getByName(value.value), executor); } catch (UnknownHostException e) { @@ -1996,11 +1997,11 @@ public class StorageService extends NotificationBroadcasterSupport implements IE } break; case SCHEMA: - SystemKeyspace.updatePeerInfo(endpoint, "schema_version", UUID.fromString(value.value)); + SystemKeyspace.updatePeerInfo(endpoint, "schema_version", UUID.fromString(value.value), executor); MigrationManager.instance.scheduleSchemaPull(endpoint, epState); break; case HOST_ID: - SystemKeyspace.updatePeerInfo(endpoint, "host_id", UUID.fromString(value.value)); + SystemKeyspace.updatePeerInfo(endpoint, "host_id", UUID.fromString(value.value), executor); break; case RPC_READY: notifyRpcChange(endpoint, epState.isRpcReady()); @@ -2046,23 +2047,24 @@ public class StorageService extends NotificationBroadcasterSupport implements IE private void updatePeerInfo(InetAddress endpoint) { EndpointState epState = Gossiper.instance.getEndpointStateForEndpoint(endpoint); + final ExecutorService executor = StageManager.getStage(Stage.MUTATION); for (Map.Entry entry : epState.states()) { switch (entry.getKey()) { case RELEASE_VERSION: - SystemKeyspace.updatePeerInfo(endpoint, "release_version", entry.getValue().value); + SystemKeyspace.updatePeerInfo(endpoint, "release_version", entry.getValue().value, executor); break; case DC: - SystemKeyspace.updatePeerInfo(endpoint, "data_center", entry.getValue().value); + SystemKeyspace.updatePeerInfo(endpoint, "data_center", entry.getValue().value, executor); break; case RACK: - SystemKeyspace.updatePeerInfo(endpoint, "rack", entry.getValue().value); + SystemKeyspace.updatePeerInfo(endpoint, "rack", entry.getValue().value, executor); break; case RPC_ADDRESS: try { - SystemKeyspace.updatePeerInfo(endpoint, "rpc_address", InetAddress.getByName(entry.getValue().value)); + SystemKeyspace.updatePeerInfo(endpoint, "rpc_address", InetAddress.getByName(entry.getValue().value), executor); } catch (UnknownHostException e) { @@ -2070,10 +2072,10 @@ public class StorageService extends NotificationBroadcasterSupport implements IE } break; case SCHEMA: - SystemKeyspace.updatePeerInfo(endpoint, "schema_version", UUID.fromString(entry.getValue().value)); + SystemKeyspace.updatePeerInfo(endpoint, "schema_version", UUID.fromString(entry.getValue().value), executor); break; case HOST_ID: - SystemKeyspace.updatePeerInfo(endpoint, "host_id", UUID.fromString(entry.getValue().value)); + SystemKeyspace.updatePeerInfo(endpoint, "host_id", UUID.fromString(entry.getValue().value), executor); break; } } @@ -2379,7 +2381,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE Gossiper.instance.replacementQuarantine(ep); // quarantine locally longer than normally; see CASSANDRA-8260 } if (!tokensToUpdateInSystemKeyspace.isEmpty()) - SystemKeyspace.updateTokens(endpoint, tokensToUpdateInSystemKeyspace); + SystemKeyspace.updateTokens(endpoint, tokensToUpdateInSystemKeyspace, StageManager.getStage(Stage.MUTATION)); if (isMoving || operationMode == Mode.MOVING) { diff --git a/test/unit/org/apache/cassandra/db/SystemKeyspaceTest.java b/test/unit/org/apache/cassandra/db/SystemKeyspaceTest.java index 256569f892..92f2a5667c 100644 --- a/test/unit/org/apache/cassandra/db/SystemKeyspaceTest.java +++ b/test/unit/org/apache/cassandra/db/SystemKeyspaceTest.java @@ -24,11 +24,14 @@ import java.net.UnknownHostException; import java.nio.file.Path; import java.nio.file.Paths; import java.util.*; +import java.util.concurrent.Future; import org.apache.commons.io.FileUtils; import org.junit.BeforeClass; import org.junit.Test; +import org.apache.cassandra.concurrent.Stage; +import org.apache.cassandra.concurrent.StageManager; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.config.SchemaConstants; import org.apache.cassandra.cql3.QueryProcessor; @@ -85,7 +88,8 @@ public class SystemKeyspaceTest { BytesToken token = new BytesToken(ByteBufferUtil.bytes("token3")); InetAddress address = InetAddress.getByName("127.0.0.2"); - SystemKeyspace.updateTokens(address, Collections.singletonList(token)); + Future future = SystemKeyspace.updateTokens(address, Collections.singletonList(token), StageManager.getStage(Stage.MUTATION)); + FBUtilities.waitOnFuture(future); assert SystemKeyspace.loadTokens().get(address).contains(token); SystemKeyspace.removeEndpoint(address); assert !SystemKeyspace.loadTokens().containsValue(token); diff --git a/test/unit/org/apache/cassandra/gms/FailureDetectorTest.java b/test/unit/org/apache/cassandra/gms/FailureDetectorTest.java index dca00b1131..0dff95a6ad 100644 --- a/test/unit/org/apache/cassandra/gms/FailureDetectorTest.java +++ b/test/unit/org/apache/cassandra/gms/FailureDetectorTest.java @@ -45,8 +45,8 @@ public class FailureDetectorTest { // slow unit tests can cause problems with FailureDetector's GC pause handling System.setProperty("cassandra.max_local_pause_in_ms", "20000"); - DatabaseDescriptor.daemonInitialization(); + DatabaseDescriptor.createAllDirectories(); } @Test diff --git a/test/unit/org/apache/cassandra/service/LeaveAndBootstrapTest.java b/test/unit/org/apache/cassandra/service/LeaveAndBootstrapTest.java index 19f0b7abc5..c5f198e3f3 100644 --- a/test/unit/org/apache/cassandra/service/LeaveAndBootstrapTest.java +++ b/test/unit/org/apache/cassandra/service/LeaveAndBootstrapTest.java @@ -22,6 +22,7 @@ package org.apache.cassandra.service; import java.net.InetAddress; import java.net.UnknownHostException; import java.util.*; +import java.util.concurrent.ExecutorService; import com.google.common.collect.HashMultimap; import com.google.common.collect.Multimap; @@ -34,6 +35,8 @@ import org.apache.cassandra.SchemaLoader; import org.apache.cassandra.Util; import org.apache.cassandra.Util.PartitionerSwitcher; import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.concurrent.Stage; +import org.apache.cassandra.concurrent.StageManager; import org.apache.cassandra.config.Schema; import org.apache.cassandra.db.SystemKeyspace; import org.apache.cassandra.dht.IPartitioner; @@ -47,6 +50,7 @@ import org.apache.cassandra.locator.AbstractReplicationStrategy; import org.apache.cassandra.locator.SimpleSnitch; import org.apache.cassandra.locator.TokenMetadata; import org.apache.cassandra.schema.KeyspaceMetadata; +import org.apache.cassandra.utils.FBUtilities; import static org.junit.Assert.*; @@ -677,8 +681,9 @@ public class LeaveAndBootstrapTest Util.createInitialRing(ss, partitioner, endpointTokens, new ArrayList(), hosts, new ArrayList(), 2); InetAddress toRemove = hosts.get(1); - SystemKeyspace.updatePeerInfo(toRemove, "data_center", "dc42"); - SystemKeyspace.updatePeerInfo(toRemove, "rack", "rack42"); + final ExecutorService executor = StageManager.getStage(Stage.MUTATION); + FBUtilities.waitOnFuture(SystemKeyspace.updatePeerInfo(toRemove, "data_center", "dc42", executor)); + FBUtilities.waitOnFuture(SystemKeyspace.updatePeerInfo(toRemove, "rack", "rack42", executor)); assertEquals("rack42", SystemKeyspace.loadDcRackInfo().get(toRemove).get("rack")); // mark the node as removed