Avoid loading the preferred IP for BulkLoader streaming

patch by Caleb Rackliffe; reviewed by Jon Meredith for CASSANDRA-18370

Co-authored-by: Caleb Rackliffe <calebrackliffe@gmail.com>
Co-authored-by: Jon Meredith <jonmeredith@apache.org>

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)

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"))

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
+     * <p> 
+     * 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;
+    }
 }

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;
         }

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
+    }
 }