diff --git a/be/src/util/hdfs-bulk-ops.cc b/be/src/util/hdfs-bulk-ops.cc
index 53d5c7a..783b4f6 100644
--- a/be/src/util/hdfs-bulk-ops.cc
+++ b/be/src/util/hdfs-bulk-ops.cc
@@ -81,8 +81,7 @@
   if (op_set_->ShouldAbort()) return;
   int err = 0;
   hdfsFS src_connection;
-  Status connection_status = HdfsFsCache::instance()->GetConnection(src_, &src_connection,
-      op_set_->connection_cache());
+  Status connection_status = op_set_->GetHdfsFsConnection(src_, &src_connection);
 
   if (!connection_status.ok()) {
     AddError(connection_status.GetDetail());
@@ -104,8 +103,7 @@
       break;
     case MOVE:
       hdfsFS dst_connection;
-      connection_status = HdfsFsCache::instance()->GetConnection(dst_, &dst_connection,
-          op_set_->connection_cache());
+      connection_status = op_set_->GetHdfsFsConnection(dst_, &dst_connection);
       if (!connection_status.ok()) break;
       err = hdfsMove(src_connection, src_.c_str(), dst_connection, dst_.c_str());
       VLOG_FILE << "hdfsMove() src_file=" << src_ << " dst_file=" << dst_;
@@ -189,3 +187,8 @@
   lock_guard<mutex> l(errors_lock_);
   return abort_on_error_ && !errors_.empty();
 }
+
+Status HdfsOperationSet::GetHdfsFsConnection(const string& path, hdfsFS* fs) {
+  lock_guard<mutex> l(connection_cache_lock_);
+  return HdfsFsCache::instance()->GetConnection(path, fs, connection_cache_);
+}
diff --git a/be/src/util/hdfs-bulk-ops.h b/be/src/util/hdfs-bulk-ops.h
index 729e5df..26ea680 100644
--- a/be/src/util/hdfs-bulk-ops.h
+++ b/be/src/util/hdfs-bulk-ops.h
@@ -128,6 +128,10 @@
   /// the set are complete.
   boost::scoped_ptr<CountingBarrier> ops_complete_barrier_;
 
+  /// Protects 'connection_cache_'. Acquired before acquiring a hdfsFS connection via
+  /// HdfsFsCache::GetConnection.
+  std::mutex connection_cache_lock_;
+
   /// A connection cache used by this operation set. Not owned.
   HdfsFsCache::HdfsFsMap* connection_cache_;
 
@@ -152,6 +156,10 @@
   /// Called by HdfsOp at the start of execution to decide whether to continue. Returns
   /// true iff abort_on_error_ is true and at least one error has been recorded.
   bool ShouldAbort();
+
+  /// Gets a hdfsFS connection for the given filesystem path. Acquires
+  /// 'connection_cache_lock_'.
+  Status GetHdfsFsConnection(const std::string& path, hdfsFS* fs);
 };
 
 }
diff --git a/bin/tsan-suppressions.txt b/bin/tsan-suppressions.txt
index 678ddf8..a93191b 100644
--- a/bin/tsan-suppressions.txt
+++ b/bin/tsan-suppressions.txt
@@ -41,3 +41,6 @@
 # 'g_kinit_ctx' (init.cc). On a production cluster, Impala daemons should only call
 # 'InitAuth' once.
 race:impala::RpcMgrKerberizedTest
+
+# TODO: IMPALA-9403: Allow TSAN to be set on codegen
+race:impala::HdfsColumnarScanner::ProcessScratchBatchCodegenOrInterpret