Further PR feedback

cpp/src/arrow/filesystem/azurefs.cc

@@ -145,6 +145,9 @@ Status AzureOptions::ExtractFromUriQuery(const Uri& uri) {
         blob_storage_scheme = "http";
         dfs_storage_scheme = "http";
       }
+    } else if (kv.first == "background_writes") {
+      ARROW_ASSIGN_OR_RAISE(background_writes,
+                            ::arrow::internal::ParseBoolean(kv.second));
     } else {
       return Status::Invalid(
           "Unexpected query parameter in Azure Blob File System URI: '", kv.first, "'");
@@ -973,6 +976,10 @@ class ObjectAppendStream final : public io::OutputStream {
  private:
   struct UploadState;
 
+  std::shared_ptr<ObjectAppendStream> Self() {
+    return std::dynamic_pointer_cast<ObjectAppendStream>(shared_from_this());
+  }
+
  public:
   ObjectAppendStream(std::shared_ptr<Blobs::BlockBlobClient> block_blob_client,
                      const io::IOContext& io_context, const AzureLocation& location,
@@ -1044,10 +1051,6 @@ class ObjectAppendStream final : public io::OutputStream {
     return Status::OK();
   }
 
-  std::shared_ptr<ObjectAppendStream> Self() {
-    return std::dynamic_pointer_cast<ObjectAppendStream>(shared_from_this());
-  }
-
   Status Abort() override {
     if (closed_) {
       return Status::OK();
@@ -1132,7 +1135,7 @@ class ObjectAppendStream final : public io::OutputStream {
   }
 
   Future<> FlushAsync() {
-    RETURN_NOT_OK(CheckClosed("flushAsync"));
+    RETURN_NOT_OK(CheckClosed("flush async"));
     if (!initialised_) {
       // If the stream has not been successfully initialized then there is nothing to
       // flush. This also avoids some unhandled errors when flushing in the destructor.

cpp/src/arrow/filesystem/azurefs_test.cc

@@ -628,6 +628,16 @@ class TestAzureOptions : public ::testing::Test {
     ASSERT_EQ(path, "container/dir/blob");
   }
 
+  void TestFromUriEnableBackgroundWrites() {
+    std::string path;
+    ASSERT_OK_AND_ASSIGN(auto options,
+                         AzureOptions::FromUri(
+                             "abfs://account:[email protected]:10000/container/dir/blob?"
+                             "background_writes=true",
+                             &path));
+    ASSERT_EQ(options.background_writes, true);
+  }
+
   void TestFromUriCredentialDefault() {
     ASSERT_OK_AND_ASSIGN(
         auto options,
@@ -771,6 +781,9 @@ TEST_F(TestAzureOptions, FromUriDfsStorage) { TestFromUriDfsStorage(); }
 TEST_F(TestAzureOptions, FromUriAbfs) { TestFromUriAbfs(); }
 TEST_F(TestAzureOptions, FromUriAbfss) { TestFromUriAbfss(); }
 TEST_F(TestAzureOptions, FromUriEnableTls) { TestFromUriEnableTls(); }
+TEST_F(TestAzureOptions, FromUriEnableBackgroundWrites) {
+  TestFromUriEnableBackgroundWrites();
+}
 TEST_F(TestAzureOptions, FromUriCredentialDefault) { TestFromUriCredentialDefault(); }
 TEST_F(TestAzureOptions, FromUriCredentialAnonymous) { TestFromUriCredentialAnonymous(); }
 TEST_F(TestAzureOptions, FromUriCredentialStorageSharedKey) {
@@ -1480,7 +1493,6 @@ class TestAzureFileSystem : public ::testing::Test {
     std::shared_ptr<Buffer> buffer;
     do {
       ASSERT_OK_AND_ASSIGN(buffer, input->Read(128 * 1024));
-      ASSERT_TRUE(buffer);
       contents.append(buffer->ToString());
     } while (buffer->size() != 0);