diff --git a/CHANGES.txt b/CHANGES.txt
index 3342318..626975b 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -25,6 +25,7 @@
  * Add guardrail for ALTER TABLE ADD / DROP / REMOVE column operations (CASSANDRA-17495)
  * Rename DisableFlag class to EnableFlag on guardrails (CASSANDRA-17544)
 Merged from 4.1:
+ * Avoid initializing schema via SystemKeyspace.getPreferredIP() with the BulkLoader tool (CASSANDRA-17740)
  * Uncomment prepared_statements_cache_size, key_cache_size, counter_cache_size, index_summary_capacity which were
    commented out by mistake in a previous patch
    Fix breaking change with cache_load_timeout; cache_load_timeout_seconds <=0 and cache_load_timeout=0 are equivalent
diff --git a/src/java/org/apache/cassandra/db/SystemKeyspace.java b/src/java/org/apache/cassandra/db/SystemKeyspace.java
index c8c21b5..4e11c93 100644
--- a/src/java/org/apache/cassandra/db/SystemKeyspace.java
+++ b/src/java/org/apache/cassandra/db/SystemKeyspace.java
@@ -42,6 +42,7 @@
 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;
@@ -988,6 +989,8 @@
      */
     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.getAddress(), ep.getPort());
         if (!result.isEmpty() && result.one().has("preferred_ip"))
diff --git a/src/java/org/apache/cassandra/streaming/StreamingChannel.java b/src/java/org/apache/cassandra/streaming/StreamingChannel.java
index a638638..6b623b8 100644
--- a/src/java/org/apache/cassandra/streaming/StreamingChannel.java
+++ b/src/java/org/apache/cassandra/streaming/StreamingChannel.java
@@ -58,6 +58,20 @@
             // Implementations can decide whether or not to do something with the preferred address.
             return create(to, messagingVersion, kind);
         }
+
+        /** 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;
+        }
     }
 
     public enum Kind { CONTROL, FILE }
diff --git a/src/java/org/apache/cassandra/streaming/async/StreamingMultiplexedChannel.java b/src/java/org/apache/cassandra/streaming/async/StreamingMultiplexedChannel.java
index c2e551e..38277e6 100644
--- a/src/java/org/apache/cassandra/streaming/async/StreamingMultiplexedChannel.java
+++ b/src/java/org/apache/cassandra/streaming/async/StreamingMultiplexedChannel.java
@@ -211,7 +211,7 @@
             if (logger.isDebugEnabled())
                 logger.debug("{} Sending {}", createLogTag(session), message);
 
-            InetAddressAndPort connectTo = SystemKeyspace.getPreferredIP(to);
+            InetAddressAndPort connectTo = factory.supportsPreferredIp() ? SystemKeyspace.getPreferredIP(to) : to;
             return fileTransferExecutor.submit(new FileStreamTask((OutgoingStreamMessage) message, connectTo));
         }
 
diff --git a/src/java/org/apache/cassandra/tools/BulkLoadConnectionFactory.java b/src/java/org/apache/cassandra/tools/BulkLoadConnectionFactory.java
index 38a9e31..b282932 100644
--- a/src/java/org/apache/cassandra/tools/BulkLoadConnectionFactory.java
+++ b/src/java/org/apache/cassandra/tools/BulkLoadConnectionFactory.java
@@ -53,7 +53,8 @@
                                    int messagingVersion,
                                    StreamingChannel.Kind kind) throws IOException
     {
-        // Supply a preferred address to the template, which will be overwritten if encryption is configured.
+        // The preferred address is always overwritten in create(). This method override only exists so we can avoid
+        // falling back to the NettyStreamingConnectionFactory implementation.
         OutboundConnectionSettings template = new OutboundConnectionSettings(getByAddress(to), getByAddress(preferred));
         return create(template, messagingVersion, kind);
     }
@@ -70,4 +71,9 @@
 
         return connect(template, messagingVersion, kind);
     }
+    @Override
+    public boolean supportsPreferredIp()
+    {
+        return false; // called in a tool context, do not use getPreferredIP
+    }
 }
diff --git a/test/unit/org/apache/cassandra/tools/ToolRunner.java b/test/unit/org/apache/cassandra/tools/ToolRunner.java
index 6f55160..d3b9f10 100644
--- a/test/unit/org/apache/cassandra/tools/ToolRunner.java
+++ b/test/unit/org/apache/cassandra/tools/ToolRunner.java
@@ -46,9 +46,7 @@
 import org.apache.cassandra.distributed.api.IInstance;
 import org.apache.cassandra.distributed.api.NodeToolResult;
 import org.apache.cassandra.utils.Pair;
-import org.assertj.core.api.Assertions;
 import org.assertj.core.util.Lists;
-import org.assertj.core.util.Strings;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.Assert.assertTrue;
@@ -402,8 +400,11 @@
          */
         public void assertCleanStdErr()
         {
-            assertTrue("Failed because cleaned stdErr wasn't empty: " + getCleanedStderr(),
-                       getCleanedStderr().isEmpty());
+            String raw = getStderr();
+            String cleaned = getCleanedStderr();
+            assertTrue("Failed to clean stderr completely.\nRaw (length=" + raw.length() + "):\n" + raw +
+                       "\nCleaned (length=" + cleaned.length() + "):\n" + cleaned,
+                       cleaned.trim().isEmpty());
         }
 
         public void assertOnExitCode()