stream analytics: add support for authentication_mode

internal/services/streamanalytics/stream_analytics_output_eventhub_resource.go

@@ -91,6 +91,16 @@ func resourceStreamAnalyticsOutputEventHub() *pluginsdk.Resource {
 				Optional: true,
 			},
 
+			"authentication_mode": {
+				Type:     pluginsdk.TypeString,
+				Optional: true,
+				Default:  string(streamanalytics.AuthenticationModeConnectionString),
+				ValidateFunc: validation.StringInSlice([]string{
+					string(streamanalytics.AuthenticationModeMsi),
+					string(streamanalytics.AuthenticationModeConnectionString),
+				}, false),
+			},
+
 			"serialization": schemaStreamAnalyticsOutputSerialization(),
 		},
 	}
@@ -141,6 +151,7 @@ func resourceStreamAnalyticsOutputEventHubCreateUpdate(d *pluginsdk.ResourceData
 					SharedAccessPolicyName: utils.String(sharedAccessPolicyName),
 					PropertyColumns:        utils.ExpandStringSlice(propertyColumns),
 					PartitionKey:           utils.String(partitionKey),
+					AuthenticationMode:     streamanalytics.AuthenticationMode(d.Get("authentication_mode").(string)),
 				},
 			},
 			Serialization: serialization,
@@ -196,6 +207,7 @@ func resourceStreamAnalyticsOutputEventHubRead(d *pluginsdk.ResourceData, meta i
 		d.Set("shared_access_policy_name", v.SharedAccessPolicyName)
 		d.Set("property_columns", v.PropertyColumns)
 		d.Set("partition_key", v.PartitionKey)
+		d.Set("authentication_mode", v.AuthenticationMode)
 
 		if err := d.Set("serialization", flattenStreamAnalyticsOutputSerialization(props.Serialization)); err != nil {
 			return fmt.Errorf("setting `serialization`: %+v", err)

internal/services/streamanalytics/stream_analytics_output_eventhub_resource_test.go

@@ -152,6 +152,21 @@ func TestAccStreamAnalyticsOutputEventHub_requiresImport(t *testing.T) {
 	})
 }
 
+func TestAccStreamAnalyticsOutputEventHub_authenticationMode(t *testing.T) {
+	data := acceptance.BuildTestData(t, "azurerm_stream_analytics_output_eventhub", "test")
+	r := StreamAnalyticsOutputEventhubResource{}
+
+	data.ResourceTest(t, r, []acceptance.TestStep{
+		{
+			Config: r.authenticationMode(data),
+			Check: acceptance.ComposeTestCheckFunc(
+				check.That(data.ResourceName).ExistsInAzure(r),
+			),
+		},
+		data.ImportStep("shared_access_policy_key"),
+	})
+}
+
 func (r StreamAnalyticsOutputEventhubResource) Exists(ctx context.Context, client *clients.Client, state *pluginsdk.InstanceState) (*bool, error) {
 	name := state.Attributes["name"]
 	jobName := state.Attributes["stream_analytics_job_name"]
@@ -365,6 +380,30 @@ resource "azurerm_stream_analytics_output_eventhub" "import" {
 `, template)
 }
 
+func (r StreamAnalyticsOutputEventhubResource) authenticationMode(data acceptance.TestData) string {
+	template := r.template(data)
+	return fmt.Sprintf(`
+%s
+
+resource "azurerm_stream_analytics_output_eventhub" "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
+  eventhub_name             = azurerm_eventhub.test.name
+  servicebus_namespace      = azurerm_eventhub_namespace.test.name
+  shared_access_policy_key  = azurerm_eventhub_namespace.test.name
+  shared_access_policy_name = "RootManageSharedAccessKey"
+  authentication_mode       = "Msi"
+
+  serialization {
+    type     = "Json"
+    encoding = "UTF8"
+    format   = "Array"
+  }
+}
+`, template, data.RandomInteger)
+}
+
 func (r StreamAnalyticsOutputEventhubResource) template(data acceptance.TestData) string {
 	return fmt.Sprintf(`
 provider "azurerm" {

internal/services/streamanalytics/stream_analytics_output_mssql_resource.go

@@ -104,6 +104,16 @@ func resourceStreamAnalyticsOutputSql() *pluginsdk.Resource {
 				Default:      1,
 				ValidateFunc: validation.FloatBetween(0, 1),
 			},
+
+			"authentication_mode": {
+				Type:     pluginsdk.TypeString,
+				Optional: true,
+				Default:  string(streamanalytics.AuthenticationModeConnectionString),
+				ValidateFunc: validation.StringInSlice([]string{
+					string(streamanalytics.AuthenticationModeMsi),
+					string(streamanalytics.AuthenticationModeConnectionString),
+				}, false),
+			},
 		},
 	}
 }
@@ -138,13 +148,14 @@ func resourceStreamAnalyticsOutputSqlCreateUpdate(d *pluginsdk.ResourceData, met
 			Datasource: &streamanalytics.AzureSQLDatabaseOutputDataSource{
 				Type: streamanalytics.TypeBasicOutputDataSourceTypeMicrosoftSQLServerDatabase,
 				AzureSQLDatabaseOutputDataSourceProperties: &streamanalytics.AzureSQLDatabaseOutputDataSourceProperties{
-					Server:         utils.String(server),
-					Database:       utils.String(databaseName),
-					User:           utils.String(sqlUser),
-					Password:       utils.String(sqlUserPassword),
-					Table:          utils.String(tableName),
-					MaxBatchCount:  utils.Float(d.Get("max_batch_count").(float64)),
-					MaxWriterCount: utils.Float(d.Get("max_writer_count").(float64)),
+					Server:             utils.String(server),
+					Database:           utils.String(databaseName),
+					User:               utils.String(sqlUser),
+					Password:           utils.String(sqlUserPassword),
+					Table:              utils.String(tableName),
+					MaxBatchCount:      utils.Float(d.Get("max_batch_count").(float64)),
+					MaxWriterCount:     utils.Float(d.Get("max_writer_count").(float64)),
+					AuthenticationMode: streamanalytics.AuthenticationMode(d.Get("authentication_mode").(string)),
 				},
 			},
 		},
@@ -198,6 +209,7 @@ func resourceStreamAnalyticsOutputSqlRead(d *pluginsdk.ResourceData, meta interf
 		d.Set("database", v.Database)
 		d.Set("table", v.Table)
 		d.Set("user", v.User)
+		d.Set("authentication_mode", v.AuthenticationMode)
 
 		maxBatchCount := float64(10000)
 		if v.MaxBatchCount != nil {

internal/services/streamanalytics/stream_analytics_output_mssql_resource_test.go

@@ -50,6 +50,21 @@ func TestAccStreamAnalyticsOutputSql_update(t *testing.T) {
 	})
 }
 
+func TestAccStreamAnalyticsOutputSql_authenticationMode(t *testing.T) {
+	data := acceptance.BuildTestData(t, "azurerm_stream_analytics_output_mssql", "test")
+	r := StreamAnalyticsOutputSqlResource{}
+
+	data.ResourceTest(t, r, []acceptance.TestStep{
+		{
+			Config: r.authenticationMode(data),
+			Check: acceptance.ComposeTestCheckFunc(
+				check.That(data.ResourceName).ExistsInAzure(r),
+			),
+		},
+		data.ImportStep("password"),
+	})
+}
+
 func TestAccStreamAnalyticsOutputSql_requiresImport(t *testing.T) {
 	data := acceptance.BuildTestData(t, "azurerm_stream_analytics_output_mssql", "test")
 	r := StreamAnalyticsOutputSqlResource{}
@@ -195,6 +210,26 @@ resource "azurerm_stream_analytics_output_mssql" "test" {
 `, template, data.RandomInteger, maxBatchCount, maxWriterCount)
 }
 
+func (r StreamAnalyticsOutputSqlResource) authenticationMode(data acceptance.TestData) string {
+	template := r.template(data)
+	return fmt.Sprintf(`
+%s
+
+resource "azurerm_stream_analytics_output_mssql" "test" {
+  name                      = "acctestoutput-%d"
+  stream_analytics_job_name = azurerm_stream_analytics_job.test.name
+  resource_group_name       = azurerm_stream_analytics_job.test.resource_group_name
+  authentication_mode       = "Msi"
+
+  server   = azurerm_sql_server.test.fully_qualified_domain_name
+  user     = azurerm_sql_server.test.administrator_login
+  password = azurerm_sql_server.test.administrator_login_password
+  database = azurerm_sql_database.test.name
+  table    = "AccTestTable"
+}
+`, template, data.RandomInteger)
+}
+
 func (r StreamAnalyticsOutputSqlResource) template(data acceptance.TestData) string {
 	return fmt.Sprintf(`
 provider "azurerm" {

internal/services/streamanalytics/stream_analytics_output_servicebus_topic_resource.go

@@ -96,6 +96,16 @@ func resourceStreamAnalyticsOutputServiceBusTopic() *pluginsdk.Resource {
 			},
 
 			"serialization": schemaStreamAnalyticsOutputSerialization(),
+
+			"authentication_mode": {
+				Type:     pluginsdk.TypeString,
+				Optional: true,
+				Default:  string(streamanalytics.AuthenticationModeConnectionString),
+				ValidateFunc: validation.StringInSlice([]string{
+					string(streamanalytics.AuthenticationModeMsi),
+					string(streamanalytics.AuthenticationModeConnectionString),
+				}, false),
+			},
 		},
 	}
 }
@@ -140,6 +150,7 @@ func resourceStreamAnalyticsOutputServiceBusTopicCreateUpdate(d *pluginsdk.Resou
 					SharedAccessPolicyName: utils.String(d.Get("shared_access_policy_name").(string)),
 					PropertyColumns:        utils.ExpandStringSlice(d.Get("property_columns").([]interface{})),
 					SystemPropertyColumns:  utils.ExpandMapStringPtrString(d.Get("system_property_columns").(map[string]interface{})),
+					AuthenticationMode:     streamanalytics.AuthenticationMode(d.Get("authentication_mode").(string)),
 				},
 			},
 			Serialization: serialization,
@@ -194,6 +205,7 @@ func resourceStreamAnalyticsOutputServiceBusTopicRead(d *pluginsdk.ResourceData,
 		d.Set("servicebus_namespace", v.ServiceBusNamespace)
 		d.Set("shared_access_policy_name", v.SharedAccessPolicyName)
 		d.Set("property_columns", v.PropertyColumns)
+		d.Set("authentication_mode", v.AuthenticationMode)
 
 		if err = d.Set("system_property_columns", utils.FlattenMapStringPtrString(v.SystemPropertyColumns)); err != nil {
 			return err

internal/services/streamanalytics/stream_analytics_output_servicebus_topic_resource_test.go

@@ -113,6 +113,21 @@ func TestAccStreamAnalyticsOutputServiceBusTopic_requiresImport(t *testing.T) {
 	})
 }
 
+func TestAccStreamAnalyticsOutputServiceBusTopic_authenticationMode(t *testing.T) {
+	data := acceptance.BuildTestData(t, "azurerm_stream_analytics_output_servicebus_topic", "test")
+	r := StreamAnalyticsOutputServiceBusTopicResource{}
+
+	data.ResourceTest(t, r, []acceptance.TestStep{
+		{
+			Config: r.json(data),
+			Check: acceptance.ComposeTestCheckFunc(
+				check.That(data.ResourceName).ExistsInAzure(r),
+			),
+		},
+		data.ImportStep("shared_access_policy_key"),
+	})
+}
+
 func TestAccStreamAnalyticsOutputServiceBusTopic_systemPropertyColumns(t *testing.T) {
 	data := acceptance.BuildTestData(t, "azurerm_stream_analytics_output_servicebus_topic", "test")
 	r := StreamAnalyticsOutputServiceBusTopicResource{}
@@ -372,6 +387,30 @@ resource "azurerm_stream_analytics_output_servicebus_topic" "import" {
 `, template)
 }
 
+func (r StreamAnalyticsOutputServiceBusTopicResource) authenticationMode(data acceptance.TestData) string {
+	template := r.template(data)
+	return fmt.Sprintf(`
+%s
+
+resource "azurerm_stream_analytics_output_servicebus_topic" "test" {
+  name                      = "acctestoutput-%d"
+  stream_analytics_job_name = azurerm_stream_analytics_job.test.name
+  resource_group_name       = azurerm_stream_analytics_job.test.resource_group_name
+  topic_name                = azurerm_servicebus_topic.test.name
+  servicebus_namespace      = azurerm_servicebus_namespace.test.name
+  shared_access_policy_key  = azurerm_servicebus_namespace.test.default_primary_key
+  shared_access_policy_name = "RootManageSharedAccessKey"
+  authentication_mode       = "Msi"
+
+  serialization {
+    type     = "Json"
+    encoding = "UTF8"
+    format   = "LineSeparated"
+  }
+}
+`, template, data.RandomInteger)
+}
+
 func (r StreamAnalyticsOutputServiceBusTopicResource) template(data acceptance.TestData) string {
 	return fmt.Sprintf(`
 provider "azurerm" {

website/docs/r/stream_analytics_output_eventhub.html.markdown

@@ -76,6 +76,8 @@ The following arguments are supported:
 
 * `property_columns` - (Optional) A list of property columns to add to the Event Hub output.
 
+* `authentication_mode` - (Optional) The authentication mode for the Stream Output. Possible values are `Msi` and `ConnectionString`. Defaults to `ConnectionString`.
+
 * `partition_key` - (Optional) The column that is used for the Event Hub partition key.
 
 ---

website/docs/r/stream_analytics_output_mssql.html.markdown

@@ -78,6 +78,8 @@ The following arguments are supported:
 
 * `max_writer_count` - (Optional) The max writer count for the SQL Database. Defaults to `1`. Possible values are `0` which bases the writer count on the query partition and `1` which corresponds to a single writer.
 
+* `authentication_mode` - (Optional) The authentication mode for the Stream Output. Possible values are `Msi` and `ConnectionString`. Defaults to `ConnectionString`.
+
 ## Attributes Reference
 
 The following attributes are exported in addition to the arguments listed above:

website/docs/r/stream_analytics_output_servicebus_topic.html.markdown

@@ -75,6 +75,8 @@ The following arguments are supported:
 
 * `property_columns` - (Optional) A list of property columns to add to the Service Bus Topic output.
 
+* `authentication_mode` - (Optional) The authentication mode for the Stream Output. Possible values are `Msi` and `ConnectionString`. Defaults to `ConnectionString`.
+
 * `system_property_columns` - (Optional) A key-value pair of system property columns that will be attached to the outgoing messages for the Service Bus Topic Output.
 
 -> **NOTE:** The acceptable keys are `ContentType`, `CorrelationId`, `Label`, `MessageId`, `PartitionKey`, `ReplyTo`, `ReplyToSessionId`, `ScheduledEnqueueTimeUtc`, `SessionId`, `TimeToLive` and `To`.