diff --git a/internal/services/streamanalytics/stream_analytics_stream_input_blob_resource.go b/internal/services/streamanalytics/stream_analytics_stream_input_blob_resource.go
index 70c1e9dd98d8..77889895c730 100644
--- a/internal/services/streamanalytics/stream_analytics_stream_input_blob_resource.go
+++ b/internal/services/streamanalytics/stream_analytics_stream_input_blob_resource.go
@@ -8,6 +8,7 @@ import (
 	"log"
 	"time"
 
+	"github.com/hashicorp/go-azure-helpers/lang/pointer"
 	"github.com/hashicorp/go-azure-helpers/lang/response"
 	"github.com/hashicorp/go-azure-helpers/resourcemanager/commonschema"
 	"github.com/hashicorp/go-azure-sdk/resource-manager/streamanalytics/2020-03-01/inputs"
@@ -98,6 +99,16 @@ func resourceStreamAnalyticsStreamInputBlob() *pluginsdk.Resource {
 			},
 
 			"serialization": schemaStreamAnalyticsStreamInputSerialization(),
+
+			"authentication_mode": {
+				Type:     pluginsdk.TypeString,
+				Optional: true,
+				Default:  string(inputs.AuthenticationModeConnectionString),
+				ValidateFunc: validation.StringInSlice([]string{
+					string(inputs.AuthenticationModeConnectionString),
+					string(inputs.AuthenticationModeMsi),
+				}, false),
+			},
 		},
 	}
 }
@@ -154,6 +165,7 @@ func resourceStreamAnalyticsStreamInputBlobCreateUpdate(d *pluginsdk.ResourceDat
 							AccountKey:  utils.String(storageAccountKey),
 						},
 					},
+					AuthenticationMode: pointer.To(inputs.AuthenticationMode(d.Get("authentication_mode").(string))),
 				},
 			},
 			Serialization: serialization,
@@ -242,6 +254,12 @@ func resourceStreamAnalyticsStreamInputBlobRead(d *pluginsdk.ResourceData, meta
 				}
 				d.Set("time_format", timeFormat)
 
+				authMode := ""
+				if v := streamBlobInputProps.AuthenticationMode; v != nil {
+					authMode = string(*v)
+				}
+				d.Set("authentication_mode", authMode)
+
 				if accounts := streamBlobInputProps.StorageAccounts; accounts != nil && len(*accounts) > 0 {
 					account := (*accounts)[0]
 					d.Set("storage_account_name", account.AccountName)
diff --git a/internal/services/streamanalytics/stream_analytics_stream_input_blob_resource_test.go b/internal/services/streamanalytics/stream_analytics_stream_input_blob_resource_test.go
index 7a3fb2a17b6f..1210f62cd464 100644
--- a/internal/services/streamanalytics/stream_analytics_stream_input_blob_resource_test.go
+++ b/internal/services/streamanalytics/stream_analytics_stream_input_blob_resource_test.go
@@ -85,6 +85,21 @@ func TestAccStreamAnalyticsStreamInputBlob_update(t *testing.T) {
 	})
 }
 
+func TestAccStreamAnalyticsStreamInputBlob_authenticationMode(t *testing.T) {
+	data := acceptance.BuildTestData(t, "azurerm_stream_analytics_stream_input_blob", "test")
+	r := StreamAnalyticsStreamInputBlobResource{}
+
+	data.ResourceTest(t, r, []acceptance.TestStep{
+		{
+			Config: r.authenticationMode(data),
+			Check: acceptance.ComposeTestCheckFunc(
+				check.That(data.ResourceName).ExistsInAzure(r),
+			),
+		},
+		data.ImportStep("storage_account_key"),
+	})
+}
+
 func TestAccStreamAnalyticsStreamInputBlob_requiresImport(t *testing.T) {
 	data := acceptance.BuildTestData(t, "azurerm_stream_analytics_stream_input_blob", "test")
 	r := StreamAnalyticsStreamInputBlobResource{}
@@ -225,6 +240,31 @@ resource "azurerm_stream_analytics_stream_input_blob" "test" {
 `, template, data.RandomString, data.RandomInteger)
 }
 
+func (r StreamAnalyticsStreamInputBlobResource) authenticationMode(data acceptance.TestData) string {
+	template := r.template(data)
+	return fmt.Sprintf(`
+%s
+
+resource "azurerm_stream_analytics_stream_input_blob" "test" {
+  name                      = "acctestinput-%d"
+  stream_analytics_job_name = azurerm_stream_analytics_job.test.name
+  resource_group_name       = azurerm_stream_analytics_job.test.resource_group_name
+  storage_account_name      = azurerm_storage_account.test.name
+  storage_account_key       = azurerm_storage_account.test.primary_access_key
+  storage_container_name    = azurerm_storage_container.test.name
+  path_pattern              = "some-random-pattern"
+  date_format               = "yyyy/MM/dd"
+  time_format               = "HH"
+  authentication_mode       = "Msi"
+
+  serialization {
+    type     = "Json"
+    encoding = "UTF8"
+  }
+}
+`, template, data.RandomInteger)
+}
+
 func (r StreamAnalyticsStreamInputBlobResource) requiresImport(data acceptance.TestData) string {
 	template := r.json(data)
 	return fmt.Sprintf(`
diff --git a/website/docs/r/stream_analytics_stream_input_blob.html.markdown b/website/docs/r/stream_analytics_stream_input_blob.html.markdown
index 98e74f58c3e0..fd43d97a1406 100644
--- a/website/docs/r/stream_analytics_stream_input_blob.html.markdown
+++ b/website/docs/r/stream_analytics_stream_input_blob.html.markdown
@@ -79,6 +79,8 @@ The following arguments are supported:
 
 * `serialization` - (Required) A `serialization` block as defined below.
 
+* `authentication_mode` - (Optional) The authentication mode for the Stream Analytics Input. Possible values are `Msi` and `ConnectionString`. Defaults to `ConnectionString`.
+
 ---
 
 A `serialization` block supports the following: