diff --git a/sdk/storage/azure-storage-blob/CHANGELOG.md b/sdk/storage/azure-storage-blob/CHANGELOG.md
index 63d2a66167804..1299838ca303e 100644
--- a/sdk/storage/azure-storage-blob/CHANGELOG.md
+++ b/sdk/storage/azure-storage-blob/CHANGELOG.md
@@ -6,6 +6,7 @@
 - Added support for 2021-12-02 service version.
 - Added support for Blob Cold Tier `AccessTier.COLD`.
 - Fixed bug where `BlobErrorCode.IncrementalCopyOfEarlierVersionSnapshotNotAllowed` was spelled incorrectly.
+- Added new overload `AppendBlobClient.getBlobOutputStream(boolean)` that takes in a boolean for overwrite and appends to existing data if overwrite is specified `false`, or deletes and recreates a blob if overwrite is specified `true`.
 
 ### Breaking Changes
 
diff --git a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/AppendBlobClient.java b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/AppendBlobClient.java
index eca6dc7dc3e39..a805f6e38e0ef 100644
--- a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/AppendBlobClient.java
+++ b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/AppendBlobClient.java
@@ -52,6 +52,7 @@
  */
 @ServiceClient(builder = SpecializedBlobClientBuilder.class)
 public final class AppendBlobClient extends BlobClientBase {
+
     private final AppendBlobAsyncClient appendBlobAsyncClient;
 
     /**
@@ -99,7 +100,7 @@ public AppendBlobClient getCustomerProvidedKeyClient(CustomerProvidedKey custome
 
     /**
      * Creates and opens an output stream to write data to the append blob. If the blob already exists on the service,
-     * it will be overwritten.
+     * new data will get appended to the existing blob.
      *
      * @return A {@link BlobOutputStream} object used to write data to the blob.
      * @throws BlobStorageException If a storage service error occurred.
@@ -109,8 +110,27 @@ public BlobOutputStream getBlobOutputStream() {
     }
 
     /**
-     * Creates and opens an output stream to write data to the append blob. If the blob already exists on the service,
-     * it will be overwritten.
+     * Creates and opens an output stream to write data to the append blob. If overwrite is specified {@code true},
+     * the existing blob will be deleted and recreated, should data exist on the blob. If overwrite is specified
+     * {@code false}, new data will get appended to the existing blob.
+     *
+     * @return A {@link BlobOutputStream} object used to write data to the blob.
+     * @param overwrite Whether an existing blob should be deleted and recreated, should data exist on the blob.
+     * @throws BlobStorageException If a storage service error occurred.
+     */
+    public BlobOutputStream getBlobOutputStream(boolean overwrite) {
+        AppendBlobRequestConditions requestConditions = null;
+        if (!overwrite) {
+            requestConditions = new AppendBlobRequestConditions().setIfNoneMatch(Constants.HeaderConstants.ETAG_WILDCARD);
+        } else {
+            // creating new blob to overwrite existing blob
+            create(true);
+        }
+        return getBlobOutputStream(requestConditions);
+    }
+
+    /**
+     * Creates and opens an output stream to write data to the append blob.
      *
      * @param requestConditions A {@link BlobRequestConditions} object that represents the access conditions for the
      * blob.
diff --git a/sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/BlobOutputStreamTest.groovy b/sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/BlobOutputStreamTest.groovy
index e79d09dcddc0d..4748ef7af8b9d 100644
--- a/sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/BlobOutputStreamTest.groovy
+++ b/sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/BlobOutputStreamTest.groovy
@@ -218,6 +218,61 @@ class BlobOutputStreamTest extends APISpec {
         convertInputStreamToByteArray(appendBlobClient.openInputStream()) == data
     }
 
+    @LiveOnly
+    def "AppendBlob output stream overwrite"() {
+        setup:
+        def data = getRandomByteArray(FOUR_MB)
+        def appendBlobClient = cc.getBlobClient(generateBlobName()).getAppendBlobClient()
+        appendBlobClient.create()
+
+        when:
+        def outputStream = appendBlobClient.getBlobOutputStream()
+        outputStream.write(data)
+        outputStream.close()
+
+        then:
+        convertInputStreamToByteArray(appendBlobClient.openInputStream()) == data
+
+        when:
+        def data2 = getRandomByteArray(FOUR_MB)
+
+        def outputStream2 = appendBlobClient.getBlobOutputStream(true)
+        outputStream2.write(data2)
+        outputStream2.close()
+
+        then:
+        appendBlobClient.getProperties().getBlobSize() == data2.length
+        convertInputStreamToByteArray(appendBlobClient.openInputStream()) == data2
+    }
+
+    @LiveOnly
+    def "AppendBlob output stream overwrite false"() {
+        setup:
+        def data = getRandomByteArray(Constants.MB)
+        def appendBlobClient = cc.getBlobClient(generateBlobName()).getAppendBlobClient()
+        appendBlobClient.create()
+
+        when:
+        def outputStream = appendBlobClient.getBlobOutputStream()
+        outputStream.write(data)
+        outputStream.close()
+
+        then:
+        convertInputStreamToByteArray(appendBlobClient.openInputStream()) == data
+
+        when:
+        def data2 = getRandomByteArray(Constants.MB)
+        outputStream = appendBlobClient.getBlobOutputStream(false)
+        outputStream.write(data2)
+        outputStream.close()
+
+        then:
+        def finalData = new byte[2 * Constants.MB]
+        System.arraycopy(data, 0, finalData, 0, data.length)
+        System.arraycopy(data2, 0, finalData, data.length, data2.length)
+        convertInputStreamToByteArray(appendBlobClient.openInputStream()) == finalData
+    }
+
     def convertInputStreamToByteArray(InputStream inputStream) {
         int b
         ByteArrayOutputStream outputStream = new ByteArrayOutputStream()