From 853ae8c84049be875921a40c9d5924724cc72792 Mon Sep 17 00:00:00 2001 From: Caleb Rackliffe Date: Thu, 30 Mar 2023 13:07:16 -0500 Subject: [PATCH] Avoid loading the preferred IP for BulkLoader streaming patch by Caleb Rackliffe; reviewed by Jon Meredith for CASSANDRA-18370 Co-authored-by: Caleb Rackliffe Co-authored-by: Jon Meredith --- CHANGES.txt | 1 + .../org/apache/cassandra/db/SystemKeyspace.java | 3 +++ .../streaming/StreamConnectionFactory.java | 14 ++++++++++++++ .../async/NettyStreamingMessageSender.java | 2 +- .../cassandra/tools/BulkLoadConnectionFactory.java | 6 ++++++ 5 files changed, 25 insertions(+), 1 deletion(-) diff --git a/CHANGES.txt b/CHANGES.txt index 675d423080..924f7d82ff 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 4.0.9 + * Avoid loading the preferred IP for BulkLoader streaming (CASSANDRA-18370) * Fix BufferPool incorrect memoryInUse when putUnusedPortion is used (CASSANDRA-18311) * Improve memtable allocator accounting when updating AtomicBTreePartition (CASSANDRA-18125) * Update zstd-jni to version 1.5.4-1 (CASSANDRA-18259) diff --git a/src/java/org/apache/cassandra/db/SystemKeyspace.java b/src/java/org/apache/cassandra/db/SystemKeyspace.java index d63ee77736..e3f7fba010 100644 --- a/src/java/org/apache/cassandra/db/SystemKeyspace.java +++ b/src/java/org/apache/cassandra/db/SystemKeyspace.java @@ -41,6 +41,7 @@ import javax.management.openmbean.OpenDataException; import javax.management.openmbean.TabularData; import com.google.common.annotations.VisibleForTesting; +import com.google.common.base.Preconditions; import com.google.common.collect.HashMultimap; import com.google.common.collect.ImmutableMap; import com.google.common.collect.ImmutableSet; @@ -934,6 +935,8 @@ public final class SystemKeyspace */ public static InetAddressAndPort getPreferredIP(InetAddressAndPort ep) { + Preconditions.checkState(DatabaseDescriptor.isDaemonInitialized()); // Make sure being used as a daemon, not a tool + String req = "SELECT preferred_ip, preferred_port FROM system.%s WHERE peer=? AND peer_port = ?"; UntypedResultSet result = executeInternal(String.format(req, PEERS_V2), ep.address, ep.port); if (!result.isEmpty() && result.one().has("preferred_ip")) diff --git a/src/java/org/apache/cassandra/streaming/StreamConnectionFactory.java b/src/java/org/apache/cassandra/streaming/StreamConnectionFactory.java index 95208e400b..97bb452272 100644 --- a/src/java/org/apache/cassandra/streaming/StreamConnectionFactory.java +++ b/src/java/org/apache/cassandra/streaming/StreamConnectionFactory.java @@ -26,4 +26,18 @@ import org.apache.cassandra.net.OutboundConnectionSettings; public interface StreamConnectionFactory { Channel createConnection(OutboundConnectionSettings template, int messagingVersion) throws IOException; + + /** Provide way to disable getPreferredIP() for tools without access to the system keyspace + *

+ * CASSANDRA-17663 moves calls to SystemKeyspace.getPreferredIP() outside of any threads + * that are regularly interrupted. However the streaming subsystem is also used + * by the bulk loader tool, which does not have direct access to the local tables + * and uses the client metadata/queries to retrieve it. + * + * @return true if SystemKeyspace.getPreferredIP() should be used when connecting + */ + default boolean supportsPreferredIp() + { + return true; + } } diff --git a/src/java/org/apache/cassandra/streaming/async/NettyStreamingMessageSender.java b/src/java/org/apache/cassandra/streaming/async/NettyStreamingMessageSender.java index 2b9111627a..9f8f47608d 100644 --- a/src/java/org/apache/cassandra/streaming/async/NettyStreamingMessageSender.java +++ b/src/java/org/apache/cassandra/streaming/async/NettyStreamingMessageSender.java @@ -238,7 +238,7 @@ public class NettyStreamingMessageSender implements StreamingMessageSender logger.debug("{} Sending {}", createLogTag(session, null), message); // Supply a preferred IP up-front to avoid trying to get it in the executor thread, which can be interrupted. - OutboundConnectionSettings templateWithConnectTo = template.withConnectTo(template.connectTo()); + OutboundConnectionSettings templateWithConnectTo = factory.supportsPreferredIp() ? template.withConnectTo(template.connectTo()) : template; fileTransferExecutor.submit(new FileStreamTask((OutgoingStreamMessage) message, templateWithConnectTo)); return; } diff --git a/src/java/org/apache/cassandra/tools/BulkLoadConnectionFactory.java b/src/java/org/apache/cassandra/tools/BulkLoadConnectionFactory.java index 04f8c66a03..5b61fabcf3 100644 --- a/src/java/org/apache/cassandra/tools/BulkLoadConnectionFactory.java +++ b/src/java/org/apache/cassandra/tools/BulkLoadConnectionFactory.java @@ -49,4 +49,10 @@ public class BulkLoadConnectionFactory extends DefaultConnectionFactory implemen return super.createConnection(template, messagingVersion); } + + @Override + public boolean supportsPreferredIp() + { + return false; // called in a tool context, do not use getPreferredIP + } }