Repository: Kimahriman/hdfs-native (language: python)

Issue:
Update object store to 0.10



Discussion / hints:

Fix patch:
diff --git a/Cargo.lock b/Cargo.lock
index 70c5e4d..79a7779 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -253,7 +253,6 @@ dependencies = [
  "iana-time-zone",
  "js-sys",
  "num-traits",
- "serde",
  "wasm-bindgen",
  "windows-targets 0.52.0",
 ]
@@ -331,16 +330,6 @@ version = "0.7.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "98cc8fbded0c607b7ba9dd60cd98df59af97e84d24e49c8557331cfc26d301ce"
 
-[[package]]
-name = "core-foundation"
-version = "0.9.4"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "91e195e091a93c46f7102ec7818a2aa394e1e1771c3ab4825963fa03e45afb8f"
-dependencies = [
- "core-foundation-sys",
- "libc",
-]
-
 [[package]]
 name = "core-foundation-sys"
 version = "0.8.6"
@@ -504,15 +493,6 @@ version = "1.10.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "11157ac094ffbdde99aa67b23417ebdd801842852b500e395a45a9c0aac03e4a"
 
-[[package]]
-name = "encoding_rs"
-version = "0.8.33"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7268b386296a025e474d5140678f75d6de9493ae55a5d709eeb9dd08149945e1"
-dependencies = [
- "cfg-if",
-]
-
 [[package]]
 name = "env_logger"
 version = "0.10.2"
@@ -554,12 +534,6 @@ version = "0.4.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0ce7134b9999ecaf8bcd65542e436736ef32ddca1b3e06094cb6ec5755203b80"
 
-[[package]]
-name = "fnv"
-version = "1.0.7"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1"
-
 [[package]]
 name = "form_urlencoded"
 version = "1.2.1"
@@ -733,25 +707,6 @@ version = "0.3.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d2fabcfbdc87f4758337ca535fb41a6d701b65693ce38287d856d1674551ec9b"
 
-[[package]]
-name = "h2"
-version = "0.3.24"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bb2c4422095b67ee78da96fbb51a4cc413b3b25883c7717ff7ca1ab31022c9c9"
-dependencies = [
- "bytes",
- "fnv",
- "futures-core",
- "futures-sink",
- "futures-util",
- "http",
- "indexmap",
- "slab",
- "tokio",
- "tokio-util",
- "tracing",
-]
-
 [[package]]
 name = "half"
 version = "2.3.1"
@@ -880,84 +835,12 @@ dependencies = [
  "windows-sys 0.52.0",
 ]
 
-[[package]]
-name = "http"
-version = "0.2.11"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8947b1a6fad4393052c7ba1f4cd97bed3e953a95c79c92ad9b051a04611d9fbb"
-dependencies = [
- "bytes",
- "fnv",
- "itoa",
-]
-
-[[package]]
-name = "http-body"
-version = "0.4.6"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7ceab25649e9960c0311ea418d17bee82c0dcec1bd053b5f9a66e265a693bed2"
-dependencies = [
- "bytes",
- "http",
- "pin-project-lite",
-]
-
-[[package]]
-name = "httparse"
-version = "1.8.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d897f394bad6a705d5f4104762e116a75639e470d80901eed05a860a95cb1904"
-
-[[package]]
-name = "httpdate"
-version = "1.0.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9"
-
 [[package]]
 name = "humantime"
 version = "2.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9a3a5bfb195931eeb336b2a7b4d761daec841b97f947d34394601737a7bba5e4"
 
-[[package]]
-name = "hyper"
-version = "0.14.28"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bf96e135eb83a2a8ddf766e426a841d8ddd7449d5f00d34ea02b41d2f19eef80"
-dependencies = [
- "bytes",
- "futures-channel",
- "futures-core",
- "futures-util",
- "h2",
- "http",
- "http-body",
- "httparse",
- "httpdate",
- "itoa",
- "pin-project-lite",
- "socket2",
- "tokio",
- "tower-service",
- "tracing",
- "want",
-]
-
-[[package]]
-name = "hyper-rustls"
-version = "0.24.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ec3efd23720e2049821a693cbc7e65ea87c72f1c58ff2f9522ff332b1491e590"
-dependencies = [
- "futures-util",
- "http",
- "hyper",
- "rustls",
- "tokio",
- "tokio-rustls",
-]
-
 [[package]]
 name = "iana-time-zone"
 version = "0.1.60"
@@ -1017,12 +900,6 @@ dependencies = [
  "generic-array",
 ]
 
-[[package]]
-name = "ipnet"
-version = "2.9.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8f518f335dce6725a761382244631d86cf0ccb2863413590b31338feb467f9c3"
-
 [[package]]
 name = "is-terminal"
 version = "0.4.12"
@@ -1173,12 +1050,6 @@ dependencies = [
  "autocfg",
 ]
 
-[[package]]
-name = "mime"
-version = "0.3.17"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a"
-
 [[package]]
 name = "minimal-lexical"
 version = "0.2.1"
@@ -1251,26 +1122,18 @@ dependencies = [
 
 [[package]]
 name = "object_store"
-version = "0.9.0"
+version = "0.10.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d139f545f64630e2e3688fd9f81c470888ab01edeb72d13b4e86c566f1130000"
+checksum = "9cda97d37376c28c508b0abc56b855c5e700324e9b753226681e658c833675c5"
 dependencies = [
  "async-trait",
- "base64",
  "bytes",
  "chrono",
  "futures",
  "humantime",
- "hyper",
  "itertools 0.12.1",
  "parking_lot",
  "percent-encoding",
- "quick-xml",
- "rand",
- "reqwest",
- "ring",
- "serde",
- "serde_json",
  "snafu",
  "tokio",
  "tracing",
@@ -1290,12 +1153,6 @@ version = "11.1.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0ab1bc2a289d34bd04a330323ac98a1b4bc82c9d9fcb1e66b63caa84da26b575"
 
-[[package]]
-name = "openssl-probe"
-version = "0.1.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ff011a302c396a5197692431fc1948019154afc178baf7d8e37367442a4601cf"
-
 [[package]]
 name = "parking_lot"
 version = "0.12.1"
@@ -1536,16 +1393,6 @@ dependencies = [
  "syn 2.0.50",
 ]
 
-[[package]]
-name = "quick-xml"
-version = "0.31.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1004a344b30a54e2ee58d66a71b32d2db2feb0a31f9a2d302bf0536f15de2a33"
-dependencies = [
- "memchr",
- "serde",
-]
-
 [[package]]
 name = "quote"
 version = "1.0.35"
@@ -1643,64 +1490,6 @@ version = "0.8.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c08c74e62047bb2de4ff487b251e4a92e24f48745648451635cec7d591162d9f"
 
-[[package]]
-name = "reqwest"
-version = "0.11.24"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c6920094eb85afde5e4a138be3f2de8bbdf28000f0029e72c45025a56b042251"
-dependencies = [
- "base64",
- "bytes",
- "encoding_rs",
- "futures-core",
- "futures-util",
- "h2",
- "http",
- "http-body",
- "hyper",
- "hyper-rustls",
- "ipnet",
- "js-sys",
- "log",
- "mime",
- "once_cell",
- "percent-encoding",
- "pin-project-lite",
- "rustls",
- "rustls-native-certs",
- "rustls-pemfile",
- "serde",
- "serde_json",
- "serde_urlencoded",
- "sync_wrapper",
- "system-configuration",
- "tokio",
- "tokio-rustls",
- "tokio-util",
- "tower-service",
- "url",
- "wasm-bindgen",
- "wasm-bindgen-futures",
- "wasm-streams",
- "web-sys",
- "winreg",
-]
-
-[[package]]
-name = "ring"
-version = "0.17.8"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c17fa4cb658e3583423e915b9f3acc01cceaee1860e33d59ebae66adc3a2dc0d"
-dependencies = [
- "cc",
- "cfg-if",
- "getrandom",
- "libc",
- "spin",
- "untrusted",
- "windows-sys 0.52.0",
-]
-
 [[package]]
 name = "roxmltree"
 version = "0.18.1"
@@ -1735,49 +1524,6 @@ dependencies = [
  "windows-sys 0.52.0",
 ]
 
-[[package]]
-name = "rustls"
-version = "0.21.10"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f9d5a6813c0759e4609cd494e8e725babae6a2ca7b62a5536a13daaec6fcb7ba"
-dependencies = [
- "log",
- "ring",
- "rustls-webpki",
- "sct",
-]
-
-[[package]]
-name = "rustls-native-certs"
-version = "0.6.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a9aace74cb666635c918e9c12bc0d348266037aa8eb599b5cba565709a8dff00"
-dependencies = [
- "openssl-probe",
- "rustls-pemfile",
- "schannel",
- "security-framework",
-]
-
-[[package]]
-name = "rustls-pemfile"
-version = "1.0.4"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1c74cae0a4cf6ccbbf5f359f08efdf8ee7e1dc532573bf0db71968cb56b1448c"
-dependencies = [
- "base64",
-]
-
-[[package]]
-name = "rustls-webpki"
-version = "0.101.7"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8b6275d1ee7a1cd780b64aca7726599a1dbc893b1e64144529e55c3c2f745765"
-dependencies = [
- "ring",
- "untrusted",
-]
-
 [[package]]
 name = "ryu"
 version = "1.0.17"
@@ -1793,54 +1539,12 @@ dependencies = [
  "winapi-util",
 ]
 
-[[package]]
-name = "schannel"
-version = "0.1.23"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fbc91545643bcf3a0bbb6569265615222618bdf33ce4ffbbd13c4bbd4c093534"
-dependencies = [
- "windows-sys 0.52.0",
-]
-
 [[package]]
 name = "scopeguard"
 version = "1.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49"
 
-[[package]]
-name = "sct"
-version = "0.7.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "da046153aa2352493d6cb7da4b6e5c0c057d8a1d0a9aa8560baffdd945acd414"
-dependencies = [
- "ring",
- "untrusted",
-]
-
-[[package]]
-name = "security-framework"
-version = "2.9.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "05b64fb303737d99b81884b2c63433e9ae28abebe5eb5045dcdd175dc2ecf4de"
-dependencies = [
- "bitflags 1.3.2",
- "core-foundation",
- "core-foundation-sys",
- "libc",
- "security-framework-sys",
-]
-
-[[package]]
-name = "security-framework-sys"
-version = "2.9.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e932934257d3b408ed8f30db49d85ea163bfe74961f017f405b025af298f0c7a"
-dependencies = [
- "core-foundation-sys",
- "libc",
-]
-
 [[package]]
 name = "serde"
 version = "1.0.197"
@@ -1872,18 +1576,6 @@ dependencies = [
  "serde",
 ]
 
-[[package]]
-name = "serde_urlencoded"
-version = "0.7.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d3491c14715ca2294c4d6a88f15e84739788c1d030eed8c110436aafdaa2f3fd"
-dependencies = [
- "form_urlencoded",
- "itoa",
- "ryu",
- "serde",
-]
-
 [[package]]
 name = "serial_test"
 version = "2.0.0"
@@ -1962,12 +1654,6 @@ dependencies = [
  "windows-sys 0.48.0",
 ]
 
-[[package]]
-name = "spin"
-version = "0.9.8"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6980e8d7511241f8acf4aebddbb1ff938df5eebe98691418c4468d0b72a96a67"
-
 [[package]]
 name = "subtle"
 version = "2.5.0"
@@ -1996,33 +1682,6 @@ dependencies = [
  "unicode-ident",
 ]
 
-[[package]]
-name = "sync_wrapper"
-version = "0.1.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2047c6ded9c721764247e62cd3b03c09ffc529b2ba5b10ec482ae507a4a70160"
-
-[[package]]
-name = "system-configuration"
-version = "0.5.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ba3a3adc5c275d719af8cb4272ea1c4a6d668a777f37e115f6d11ddbc1c8e0e7"
-dependencies = [
- "bitflags 1.3.2",
- "core-foundation",
- "system-configuration-sys",
-]
-
-[[package]]
-name = "system-configuration-sys"
-version = "0.5.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a75fb188eb626b924683e3b95e3a48e63551fcfb51949de2f06a9d91dbee93c9"
-dependencies = [
- "core-foundation-sys",
- "libc",
-]
-
 [[package]]
 name = "target-lexicon"
 version = "0.12.13"
@@ -2123,36 +1782,6 @@ dependencies = [
  "syn 2.0.50",
 ]
 
-[[package]]
-name = "tokio-rustls"
-version = "0.24.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c28327cf380ac148141087fbfb9de9d7bd4e84ab5d2c28fbc911d753de8a7081"
-dependencies = [
- "rustls",
- "tokio",
-]
-
-[[package]]
-name = "tokio-util"
-version = "0.7.10"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5419f34732d9eb6ee4c3578b7989078579b7f039cbbb9ca2c4da015749371e15"
-dependencies = [
- "bytes",
- "futures-core",
- "futures-sink",
- "pin-project-lite",
- "tokio",
- "tracing",
-]
-
-[[package]]
-name = "tower-service"
-version = "0.3.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b6bc1c9ce2b5135ac7f93c72918fc37feb872bdc6a5533a8b85eb4b86bfdae52"
-
 [[package]]
 name = "tracing"
 version = "0.1.40"
@@ -2184,12 +1813,6 @@ dependencies = [
  "once_cell",
 ]
 
-[[package]]
-name = "try-lock"
-version = "0.2.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b"
-
 [[package]]
 name = "typenum"
 version = "1.17.0"
@@ -2223,12 +1846,6 @@ version = "0.2.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c7de7d73e1754487cb58364ee906a499937a0dfabd86bcb980fa99ec8c8fa2ce"
 
-[[package]]
-name = "untrusted"
-version = "0.9.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1"
-
 [[package]]
 name = "url"
 version = "2.5.0"
@@ -2274,15 +1891,6 @@ dependencies = [
  "winapi-util",
 ]
 
-[[package]]
-name = "want"
-version = "0.3.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bfa7760aed19e106de2c7c0b581b509f2f25d3dacaf737cb82ac61bc6d760b0e"
-dependencies = [
- "try-lock",
-]
-
 [[package]]
 name = "wasi"
 version = "0.11.0+wasi-snapshot-preview1"
@@ -2314,18 +1922,6 @@ dependencies = [
  "wasm-bindgen-shared",
 ]
 
-[[package]]
-name = "wasm-bindgen-futures"
-version = "0.4.41"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "877b9c3f61ceea0e56331985743b13f3d25c406a7098d45180fb5f09bc19ed97"
-dependencies = [
- "cfg-if",
- "js-sys",
- "wasm-bindgen",
- "web-sys",
-]
-
 [[package]]
 name = "wasm-bindgen-macro"
 version = "0.2.91"
@@ -2355,19 +1951,6 @@ version = "0.2.91"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4f186bd2dcf04330886ce82d6f33dd75a7bfcf69ecf5763b89fcde53b6ac9838"
 
-[[package]]
-name = "wasm-streams"
-version = "0.4.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b65dc4c90b63b118468cf747d8bf3566c1913ef60be765b5730ead9e0a3ba129"
-dependencies = [
- "futures-util",
- "js-sys",
- "wasm-bindgen",
- "wasm-bindgen-futures",
- "web-sys",
-]
-
 [[package]]
 name = "web-sys"
 version = "0.3.68"
@@ -2562,16 +2145,6 @@ version = "0.52.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "dff9641d1cd4be8d1a070daf9e3773c5f67e78b4d9d42263020c057706765c04"
 
-[[package]]
-name = "winreg"
-version = "0.50.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "524e57b2c537c0f9b1e69f1965311ec12182b4122e45035b1508cd24d2adadb1"
-dependencies = [
- "cfg-if",
- "windows-sys 0.48.0",
-]
-
 [[package]]
 name = "xmlparser"
 version = "0.13.6"
diff --git a/crates/hdfs-native-object-store/Cargo.toml b/crates/hdfs-native-object-store/Cargo.toml
index a47d336..de76d8d 100644
--- a/crates/hdfs-native-object-store/Cargo.toml
+++ b/crates/hdfs-native-object-store/Cargo.toml
@@ -16,7 +16,7 @@ bytes = { workspace = true }
 chrono = { workspace = true }
 futures = { workspace = true }
 hdfs-native = { path = "../hdfs-native", version = "0.9" }
-object_store = { version = "0.9", features = ["cloud"] }
+object_store = { version = "0.10" }
 thiserror = "1"
 tokio = { workspace = true, features = ["rt", "net", "io-util", "macros", "sync", "time"] }
 
diff --git a/crates/hdfs-native-object-store/src/lib.rs b/crates/hdfs-native-object-store/src/lib.rs
index 6f3e649..e52132a 100644
--- a/crates/hdfs-native-object-store/src/lib.rs
+++ b/crates/hdfs-native-object-store/src/lib.rs
@@ -4,12 +4,10 @@
 //! # Usage
 //!
 //! ```rust
-//! use hdfs_native::Client;
 //! use hdfs_native_object_store::HdfsObjectStore;
-//! # use hdfs_native::Result;
+//! # use object_store::Result;
 //! # fn main() -> Result<()> {
-//! let client = Client::new("hdfs://localhost:9000")?;
-//! let store = HdfsObjectStore::new(client);
+//! let store = HdfsObjectStore::with_url("hdfs://localhost:9000")?;
 //! # Ok(())
 //! # }
 //! ```
@@ -17,24 +15,25 @@ use std::{
     fmt::{Display, Formatter},
     future,
     path::PathBuf,
-    sync::{
-        atomic::{AtomicUsize, Ordering},
-        Arc,
-    },
+    sync::Arc,
 };
 
 use async_trait::async_trait;
-use bytes::Bytes;
 use chrono::{NaiveDateTime, TimeZone, Utc};
-use futures::stream::{BoxStream, StreamExt};
+use futures::{
+    stream::{BoxStream, StreamExt},
+    FutureExt,
+};
 use hdfs_native::{client::FileStatus, file::FileWriter, Client, HdfsError, WriteOptions};
 use object_store::{
-    multipart::{PartId, PutPart, WriteMultiPart},
-    path::Path,
-    GetOptions, GetRange, GetResult, GetResultPayload, ListResult, MultipartId, ObjectMeta,
-    ObjectStore, PutMode, PutOptions, PutResult, Result,
+    path::Path, GetOptions, GetRange, GetResult, GetResultPayload, ListResult, MultipartUpload,
+    ObjectMeta, ObjectStore, PutMode, PutMultipartOpts, PutOptions, PutPayload, PutResult, Result,
+    UploadPart,
+};
+use tokio::{
+    sync::{mpsc, oneshot},
+    task::{self, JoinHandle},
 };
-use tokio::io::AsyncWrite;
 
 #[derive(Debug)]
 pub struct HdfsObjectStore {
@@ -42,10 +41,14 @@ pub struct HdfsObjectStore {
 }
 
 impl HdfsObjectStore {
-    pub fn new(client: Client) -> Self {
-        Self {
-            client: Arc::new(client),
-        }
+    pub fn new(client: Arc<Client>) -> Self {
+        Self { client }
+    }
+
+    pub fn with_url(url: &str) -> Result<Self> {
+        Ok(Self {
+            client: Arc::new(Client::new(url).to_object_store_err()?),
+        })
     }
 
     async fn internal_copy(&self, from: &Path, to: &Path, overwrite: bool) -> Result<()> {
@@ -81,6 +84,37 @@ impl HdfsObjectStore {
 
         Ok(())
     }
+
+    async fn open_tmp_file(&self, file_path: &str) -> Result<(FileWriter, String)> {
+        let path_buf = PathBuf::from(file_path);
+
+        let file_name = path_buf
+            .file_name()
+            .ok_or(HdfsError::InvalidPath("path missing filename".to_string()))
+            .to_object_store_err()?
+            .to_str()
+            .ok_or(HdfsError::InvalidPath("path not valid unicode".to_string()))
+            .to_object_store_err()?
+            .to_string();
+
+        let tmp_file_path = path_buf
+            .with_file_name(format!(".{}.tmp", file_name))
+            .to_str()
+            .ok_or(HdfsError::InvalidPath("path not valid unicode".to_string()))
+            .to_object_store_err()?
+            .to_string();
+
+        // Try to create a file with an incrementing index until we find one that doesn't exist yet
+        let mut index = 1;
+        loop {
+            let path = format!("{}.{}", tmp_file_path, index);
+            match self.client.create(&path, WriteOptions::default()).await {
+                Ok(writer) => break Ok((writer, path)),
+                Err(HdfsError::AlreadyExists(_)) => index += 1,
+                Err(e) => break Err(e).to_object_store_err(),
+            }
+        }
+    }
 }
 
 impl Display for HdfsObjectStore {
@@ -91,7 +125,7 @@ impl Display for HdfsObjectStore {
 
 impl From<Client> for HdfsObjectStore {
     fn from(value: Client) -> Self {
-        Self::new(value)
+        Self::new(Arc::new(value))
     }
 }
 
@@ -99,9 +133,14 @@ impl From<Client> for HdfsObjectStore {
 impl ObjectStore for HdfsObjectStore {
     /// Save the provided bytes to the specified location
     ///
-    /// To make the operation atomic, we write to a temporary file ".{filename}.tmp" and rename
+    /// To make the operation atomic, we write to a temporary file ".{filename}.tmp.{i}" and rename
     /// on a successful write.
-    async fn put_opts(&self, location: &Path, bytes: Bytes, opts: PutOptions) -> Result<PutResult> {
+    async fn put_opts(
+        &self,
+        location: &Path,
+        payload: PutPayload,
+        opts: PutOptions,
+    ) -> Result<PutResult> {
         let overwrite = match opts.mode {
             PutMode::Create => false,
             PutMode::Overwrite => true,
@@ -113,39 +152,23 @@ impl ObjectStore for HdfsObjectStore {
         };
 
         let final_file_path = make_absolute_file(location);
-        let path_buf = PathBuf::from(&final_file_path);
-
-        let file_name = path_buf
-            .file_name()
-            .ok_or(HdfsError::InvalidPath("path missing filename".to_string()))
-            .to_object_store_err()?
-            .to_str()
-            .ok_or(HdfsError::InvalidPath("path not valid unicode".to_string()))
-            .to_object_store_err()?
-            .to_string();
 
-        let tmp_filename = path_buf
-            .with_file_name(format!(".{}.tmp", file_name))
-            .to_str()
-            .ok_or(HdfsError::InvalidPath("path not valid unicode".to_string()))
-            .to_object_store_err()?
-            .to_string();
+        // If we're not overwriting, do an upfront check to see if the file already
+        // exists. Otherwise we have to write the whole file and try to rename before
+        // finding out.
+        if !overwrite && self.client.get_file_info(&final_file_path).await.is_ok() {
+            return Err(HdfsError::AlreadyExists(final_file_path)).to_object_store_err();
+        }
 
-        let write_options = WriteOptions {
-            overwrite,
-            ..Default::default()
-        };
+        let (mut tmp_file, tmp_file_path) = self.open_tmp_file(&final_file_path).await?;
 
-        let mut writer = self
-            .client
-            .create(&tmp_filename, write_options)
-            .await
-            .to_object_store_err()?;
-        writer.write(bytes).await.to_object_store_err()?;
-        writer.close().await.to_object_store_err()?;
+        for bytes in payload {
+            tmp_file.write(bytes).await.to_object_store_err()?;
+        }
+        tmp_file.close().await.to_object_store_err()?;
 
         self.client
-            .rename(&tmp_filename, &final_file_path, overwrite)
+            .rename(&tmp_file_path, &final_file_path, overwrite)
             .await
             .to_object_store_err()?;
 
@@ -155,71 +178,23 @@ impl ObjectStore for HdfsObjectStore {
         })
     }
 
-    /// Uses the [PutPart] trait to implement an asynchronous writer. We can't actually upload
-    /// multiple parts at once, so we simply set a limit of one part at a time.
-    async fn put_multipart(
+    /// Create a multipart writer that writes to a temporary file in a background task, and renames
+    /// to the final destination on complete.
+    async fn put_multipart_opts(
         &self,
         location: &Path,
-    ) -> Result<(MultipartId, Box<dyn AsyncWrite + Unpin + Send>)> {
+        _opts: PutMultipartOpts,
+    ) -> Result<Box<dyn MultipartUpload>> {
         let final_file_path = make_absolute_file(location);
-        let path_buf = PathBuf::from(&final_file_path);
 
-        let file_name = path_buf
-            .file_name()
-            .ok_or(HdfsError::InvalidPath("path missing filename".to_string()))
-            .to_object_store_err()?
-            .to_str()
-            .ok_or(HdfsError::InvalidPath("path not valid unicode".to_string()))
-            .to_object_store_err()?
-            .to_string();
+        let (tmp_file, tmp_file_path) = self.open_tmp_file(&final_file_path).await?;
 
-        let tmp_filename = path_buf
-            .with_file_name(format!(".{}.tmp", file_name))
-            .to_str()
-            .ok_or(HdfsError::InvalidPath("path not valid unicode".to_string()))
-            .to_object_store_err()?
-            .to_string();
-
-        // First we need to check if the tmp file exists so we know whether to overwrite
-        let overwrite = match self.client.get_file_info(&tmp_filename).await {
-            Ok(_) => true,
-            Err(HdfsError::FileNotFound(_)) => false,
-            Err(e) => Err(e).to_object_store_err()?,
-        };
-
-        let write_options = WriteOptions {
-            overwrite,
-            ..Default::default()
-        };
-
-        let writer = self
-            .client
-            .create(&tmp_filename, write_options)
-            .await
-            .to_object_store_err()?;
-
-        Ok((
-            tmp_filename.clone(),
-            Box::new(WriteMultiPart::new(
-                HdfsMultipartWriter::new(
-                    Arc::clone(&self.client),
-                    writer,
-                    &tmp_filename,
-                    &final_file_path,
-                ),
-                1,
-            )),
-        ))
-    }
-
-    /// Attempts to delete the temporary file used for multipart uploads.
-    async fn abort_multipart(&self, _location: &Path, multipart_id: &MultipartId) -> Result<()> {
-        // The multipart_id is the resolved temporary file name, so we can just delete it
-        self.client
-            .delete(multipart_id, false)
-            .await
-            .to_object_store_err()?;
-        Ok(())
+        Ok(Box::new(HdfsMultipartWriter::new(
+            Arc::clone(&self.client),
+            tmp_file,
+            &tmp_file_path,
+            &final_file_path,
+        )))
     }
 
     async fn get_opts(&self, location: &Path, options: GetOptions) -> Result<GetResult> {
@@ -258,6 +233,7 @@ impl ObjectStore for HdfsObjectStore {
             payload,
             meta,
             range,
+            attributes: Default::default(),
         })
     }
 
@@ -434,75 +410,148 @@ impl<T> HdfsErrorConvert<T> for hdfs_native::Result<T> {
     }
 }
 
-// Create a fake multipart writer that assumes only one part will be
-// written at a time. It would be better if we figured out how to implement
-// AsyncWrite for the FileWriter
+type PartSender = mpsc::UnboundedSender<(oneshot::Sender<Result<()>>, PutPayload)>;
+
+// Create a fake multipart writer the creates an uploader to a temp file as a background
+// task, and submits new parts to be uploaded to a queue for this task.
+// A once cell is used to track whether a part has finished writing or not.
+// On completing, rename the file to the actual target.
 struct HdfsMultipartWriter {
-    // FileWriter is stateful, but put_part doesn't allow a mutable borrow so we
-    // have to wrap in an async mutex
     client: Arc<Client>,
-    inner: Arc<tokio::sync::Mutex<FileWriter>>,
+    sender: Option<(JoinHandle<Result<()>>, PartSender)>,
     tmp_filename: String,
     final_filename: String,
-    next_part: AtomicUsize,
 }
 
 impl HdfsMultipartWriter {
     fn new(
         client: Arc<Client>,
-        inner: FileWriter,
+        writer: FileWriter,
         tmp_filename: &str,
         final_filename: &str,
     ) -> Self {
+        let (sender, receiver) = mpsc::unbounded_channel();
+
         Self {
             client,
-            inner: Arc::new(tokio::sync::Mutex::new(inner)),
+            sender: Some((Self::start_writer_task(writer, receiver), sender)),
             tmp_filename: tmp_filename.to_string(),
             final_filename: final_filename.to_string(),
-            next_part: AtomicUsize::new(0),
         }
     }
+
+    fn start_writer_task(
+        mut writer: FileWriter,
+        mut part_receiver: mpsc::UnboundedReceiver<(oneshot::Sender<Result<()>>, PutPayload)>,
+    ) -> JoinHandle<Result<()>> {
+        task::spawn(async move {
+            'outer: loop {
+                match part_receiver.recv().await {
+                    Some((sender, part)) => {
+                        for bytes in part {
+                            if let Err(e) = writer.write(bytes).await.to_object_store_err() {
+                                let _ = sender.send(Err(e));
+                                break 'outer;
+                            }
+                        }
+                        let _ = sender.send(Ok(()));
+                    }
+                    None => {
+                        return writer.close().await.to_object_store_err();
+                    }
+                }
+            }
+
+            // If we've reached here, a write task failed so just return Err's for all new parts that come in
+            while let Some((sender, _)) = part_receiver.recv().await {
+                let _ = sender.send(
+                    Err(HdfsError::OperationFailed(
+                        "Write failed during one of the parts".to_string(),
+                    ))
+                    .to_object_store_err(),
+                );
+            }
+            Err(HdfsError::OperationFailed(
+                "Write failed during one of the parts".to_string(),
+            ))
+            .to_object_store_err()
+        })
+    }
+}
+
+impl std::fmt::Debug for HdfsMultipartWriter {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("HdfsMultipartWriter")
+            .field("tmp_filename", &self.tmp_filename)
+            .field("final_filename", &self.final_filename)
+            .finish()
+    }
 }
 
 #[async_trait]
-impl PutPart for HdfsMultipartWriter {
-    /// Upload a single part
-    async fn put_part(&self, buf: Vec<u8>, part_idx: usize) -> Result<PartId> {
-        if part_idx != self.next_part.load(Ordering::SeqCst) {
-            return Err(object_store::Error::NotSupported {
-                source: "Part received out of order".to_string().into(),
-            });
+impl MultipartUpload for HdfsMultipartWriter {
+    fn put_part(&mut self, payload: PutPayload) -> UploadPart {
+        let (result_sender, result_receiver) = oneshot::channel();
+
+        if let Some((_, payload_sender)) = self.sender.as_ref() {
+            payload_sender.send((result_sender, payload)).unwrap();
+        } else {
+            result_sender
+                .send(
+                    Err(HdfsError::OperationFailed(
+                        "Cannot put part after completing or aborting".to_string(),
+                    ))
+                    .to_object_store_err(),
+                )
+                .unwrap();
         }
 
-        self.inner
-            .lock()
-            .await
-            .write(buf.into())
-            .await
-            .to_object_store_err()?;
+        async { result_receiver.await.unwrap() }.boxed()
+    }
 
-        self.next_part.fetch_add(1, Ordering::SeqCst);
+    async fn complete(&mut self) -> Result<PutResult> {
+        // Drop the sender so the task knows no more data is coming
+        if let Some((handle, sender)) = self.sender.take() {
+            drop(sender);
 
-        Ok(PartId {
-            content_id: part_idx.to_string(),
-        })
+            // Wait for the writer task to finish
+            handle.await??;
+
+            self.client
+                .rename(&self.tmp_filename, &self.final_filename, true)
+                .await
+                .to_object_store_err()?;
+
+            Ok(PutResult {
+                e_tag: None,
+                version: None,
+            })
+        } else {
+            Err(object_store::Error::NotSupported {
+                source: "Cannot call abort or complete multiple times".into(),
+            })
+        }
     }
 
-    /// Complete the upload with the provided parts
-    ///
-    /// `completed_parts` is in order of part number
-    async fn complete(&self, _completed_parts: Vec<PartId>) -> Result<()> {
-        self.inner
-            .lock()
-            .await
-            .close()
-            .await
-            .to_object_store_err()?;
-        self.client
-            .rename(&self.tmp_filename, &self.final_filename, true)
-            .await
-            .to_object_store_err()?;
-        Ok(())
+    async fn abort(&mut self) -> Result<()> {
+        // Drop the sender so the task knows no more data is coming
+        if let Some((handle, sender)) = self.sender.take() {
+            drop(sender);
+
+            // Wait for the writer task to finish
+            handle.abort();
+
+            self.client
+                .delete(&self.tmp_filename, false)
+                .await
+                .to_object_store_err()?;
+
+            Ok(())
+        } else {
+            Err(object_store::Error::NotSupported {
+                source: "Cannot call abort or complete multiple times".into(),
+            })
+        }
     }
 }
 
diff --git a/crates/hdfs-native/src/hdfs/proxy.rs b/crates/hdfs-native/src/hdfs/proxy.rs
index d751c52..b69dd80 100644
--- a/crates/hdfs-native/src/hdfs/proxy.rs
+++ b/crates/hdfs-native/src/hdfs/proxy.rs
@@ -171,6 +171,9 @@ impl NameServiceProxy {
     fn convert_rpc_error(exception: String, msg: String) -> HdfsError {
         match exception.as_ref() {
             "org.apache.hadoop.fs.FileAlreadyExistsException" => HdfsError::AlreadyExists(msg),
+            "org.apache.hadoop.hdfs.protocol.AlreadyBeingCreatedException" => {
+                HdfsError::AlreadyExists(msg)
+            }
             _ => HdfsError::RPCError(exception, msg),
         }
     }


Test patch:
diff --git a/crates/hdfs-native-object-store/tests/common/mod.rs b/crates/hdfs-native-object-store/tests/common/mod.rs
deleted file mode 100644
index 9969a89..0000000
--- a/crates/hdfs-native-object-store/tests/common/mod.rs
+++ /dev/null
@@ -1,40 +0,0 @@
-#![allow(dead_code)]
-use std::collections::HashSet;
-use std::io::{BufWriter, Write};
-use std::process::Command;
-use tempfile::NamedTempFile;
-use which::which;
-
-use hdfs_native::minidfs::{DfsFeatures, MiniDfs};
-
-pub const TEST_FILE_INTS: usize = 64 * 1024 * 1024;
-
-pub fn setup(features: &HashSet<DfsFeatures>) -> MiniDfs {
-    let hadoop_exc = which("hadoop").expect("Failed to find hadoop executable");
-
-    let dfs = MiniDfs::with_features(features);
-
-    let mut file = NamedTempFile::new_in("target/test").unwrap();
-    {
-        let mut writer = BufWriter::new(file.as_file_mut());
-        for i in 0..TEST_FILE_INTS as i32 {
-            let bytes = i.to_be_bytes();
-            writer.write_all(&bytes).unwrap();
-        }
-        writer.flush().unwrap();
-    }
-
-    let status = Command::new(hadoop_exc)
-        .args([
-            "fs",
-            "-copyFromLocal",
-            "-f",
-            file.path().to_str().unwrap(),
-            &format!("{}/testfile", dfs.url),
-        ])
-        .status()
-        .unwrap();
-    assert!(status.success());
-
-    dfs
-}
diff --git a/crates/hdfs-native-object-store/tests/test_object_store.rs b/crates/hdfs-native-object-store/tests/test_object_store.rs
index 5849f29..805bc5c 100644
--- a/crates/hdfs-native-object-store/tests/test_object_store.rs
+++ b/crates/hdfs-native-object-store/tests/test_object_store.rs
@@ -1,23 +1,36 @@
-#[cfg(feature = "integration-test")]
-mod common;
-
 #[cfg(feature = "integration-test")]
 mod test {
-    use bytes::{Buf, BufMut, Bytes, BytesMut};
-    use hdfs_native::{minidfs::DfsFeatures, Client};
+    use bytes::{Buf, BufMut, BytesMut};
+    use hdfs_native::{
+        minidfs::{DfsFeatures, MiniDfs},
+        Client, WriteOptions,
+    };
     use hdfs_native_object_store::{HdfsErrorConvert, HdfsObjectStore};
-    use object_store::{PutMode, PutOptions};
+    use object_store::{PutMode, PutOptions, PutPayload};
     use serial_test::serial;
-    use std::collections::HashSet;
+    use std::{collections::HashSet, sync::Arc};
 
-    use crate::common::{setup, TEST_FILE_INTS};
+    pub const TEST_FILE_INTS: usize = 64 * 1024 * 1024;
 
     #[tokio::test]
     #[serial]
     async fn test_object_store() -> object_store::Result<()> {
-        let dfs = setup(&HashSet::from([DfsFeatures::HA]));
+        let dfs = MiniDfs::with_features(&HashSet::from([DfsFeatures::HA]));
         let client = Client::new(&dfs.url).to_object_store_err()?;
-        let store = HdfsObjectStore::new(client);
+
+        // Create a test file with the client directly to sanity check reads and lists
+        let mut file = client
+            .create("/testfile", WriteOptions::default())
+            .await
+            .unwrap();
+        let mut buf = BytesMut::new();
+        for i in 0..TEST_FILE_INTS as i32 {
+            buf.put_i32(i);
+        }
+        file.write(buf.freeze()).await.unwrap();
+        file.close().await.unwrap();
+
+        let store = HdfsObjectStore::new(Arc::new(client));
 
         test_object_store_head(&store).await?;
         test_object_store_list(&store).await?;
@@ -108,14 +121,16 @@ mod test {
     async fn test_object_store_write(store: &HdfsObjectStore) -> object_store::Result<()> {
         use object_store::{path::Path, ObjectStore};
 
-        store.put(&Path::from("/newfile"), Bytes::new()).await?;
+        store
+            .put(&Path::from("/newfile"), PutPayload::new())
+            .await?;
         store.head(&Path::from("/newfile")).await?;
 
         // PutMode = Create should fail for existing file
         match store
             .put_opts(
                 &Path::from("/newfile"),
-                Bytes::new(),
+                PutPayload::new(),
                 PutOptions {
                     mode: PutMode::Create,
                     ..Default::default()
@@ -138,7 +153,9 @@ mod test {
             }
 
             let buf = data.freeze();
-            store.put(&Path::from("/newfile"), buf.clone()).await?;
+            store
+                .put(&Path::from("/newfile"), PutPayload::from_bytes(buf.clone()))
+                .await?;
 
             assert_eq!(
                 store.head(&Path::from("/newfile")).await?.size,
@@ -166,25 +183,18 @@ mod test {
     async fn test_object_store_write_multipart(
         store: &HdfsObjectStore,
     ) -> object_store::Result<()> {
-        use hdfs_native::HdfsError;
         use object_store::{path::Path, ObjectStore};
-        use tokio::io::AsyncWriteExt;
 
-        let (_, mut writer) = store.put_multipart(&"/newfile".into()).await?;
-        writer
-            .shutdown()
-            .await
-            .map_err(HdfsError::from)
-            .to_object_store_err()?;
+        let mut uploader = store.put_multipart(&"/newfile".into()).await?;
+        uploader.complete().await?;
 
-        store.put(&Path::from("/newfile"), Bytes::new()).await?;
         store.head(&Path::from("/newfile")).await?;
 
         // Check a small files, a file that is exactly one block, and a file slightly bigger than a block
         for size_to_check in [16i32, 128 * 1024 * 1024, 130 * 1024 * 1024] {
             let ints_to_write = size_to_check / 4;
 
-            let (_, mut writer) = store.put_multipart(&"/newfile".into()).await?;
+            let mut uploader = store.put_multipart(&"/newfile".into()).await?;
 
             let mut data = BytesMut::with_capacity(size_to_check as usize);
             for i in 0..ints_to_write {
@@ -195,18 +205,10 @@ mod test {
             let mut buf = data.freeze();
             while !buf.is_empty() {
                 let to_write = usize::min(buf.len(), 10 * 1024 * 1024);
-                writer
-                    .write_all_buf(&mut buf.split_to(to_write))
-                    .await
-                    .map_err(HdfsError::from)
-                    .to_object_store_err()?;
+                uploader.put_part(buf.split_to(to_write).into()).await?;
             }
 
-            writer
-                .shutdown()
-                .await
-                .map_err(HdfsError::from)
-                .to_object_store_err()?;
+            uploader.complete().await?;
 
             assert_eq!(
                 store.head(&Path::from("/newfile")).await?.size,
@@ -231,12 +233,22 @@ mod test {
         store.delete(&Path::from("/newfile")).await?;
 
         // Test aborting
-        let (multipart_id, _) = store.put_multipart(&"/newfile".into()).await?;
-        assert!(store.head(&"/.newfile.tmp".into()).await.is_ok());
-        store
-            .abort_multipart(&"/newfile".into(), &multipart_id)
-            .await?;
-        assert!(store.head(&"/.newfile.tmp".into()).await.is_err());
+        let mut uploader = store.put_multipart(&"/newfile".into()).await?;
+        assert!(store.head(&"/.newfile.tmp.1".into()).await.is_ok());
+        uploader.abort().await?;
+        assert!(store.head(&"/.newfile.tmp.1".into()).await.is_err());
+        assert!(store.head(&"/newfile".into()).await.is_err());
+
+        // Test multiple uploads to the same destination, default is to overwrite
+        let mut uploader1 = store.put_multipart(&"/newfile".into()).await?;
+        let mut uploader2 = store.put_multipart(&"/newfile".into()).await?;
+        uploader1.put_part(vec![1].into()).await?;
+        uploader2.put_part(vec![2].into()).await?;
+        uploader1.complete().await?;
+        uploader2.complete().await?;
+
+        let result = store.get(&"/newfile".into()).await?;
+        assert!(result.bytes().await?.to_vec() == vec![2]);
 
         Ok(())
     }
@@ -244,7 +256,9 @@ mod test {
     async fn test_object_store_copy(store: &HdfsObjectStore) -> object_store::Result<()> {
         use object_store::{path::Path, ObjectStore};
 
-        store.put(&Path::from("/newfile"), Bytes::new()).await?;
+        store
+            .put(&Path::from("/newfile"), PutPayload::new())
+            .await?;
 
         let size_to_check = 128 * 1024 * 1024;
         let ints_to_write = size_to_check / 4;
@@ -254,7 +268,9 @@ mod test {
         }
 
         let buf = data.freeze();
-        store.put(&Path::from("/newfile"), buf.clone()).await?;
+        store
+            .put(&Path::from("/newfile"), buf.clone().into())
+            .await?;
         store
             .copy(&Path::from("/newfile"), &Path::from("/newfile2"))
             .await?;