Add possibility to create EventHub namespace with kafka

azure-mgmt-eventhub/src/main/java/com/microsoft/azure/management/eventhub/EventHubNamespace.java

@@ -64,6 +64,11 @@ public interface EventHubNamespace extends
      */
     @Beta(Beta.SinceVersion.V1_7_0)
     boolean isAutoScaleEnabled();
+    /**
+     * @return true if kafka enabled for the namespace, false otherwise
+     */
+    @Beta(Beta.SinceVersion.V1_7_0)
+    boolean isKafkaEnabled();
     /**
      * @return current throughput units set for the namespace
      */
@@ -138,6 +143,19 @@ interface WithSku {
             WithCreate withSku(EventHubNamespaceSkuType namespaceSku);
         }
 
+        /**
+         * The stage of the event hub namespace definition enabling kafka.
+         */
+        @Beta(Beta.SinceVersion.V1_7_0)
+        interface WithKafka {
+            /**
+             * Enables kafka.
+             *
+             * @return next stage of the event hub namespace definition
+             */
+            WithCreate withKafka();
+        }
+
         /**
          * The stage of the event hub namespace definition allowing to add new event hub in the namespace.
          */
@@ -247,7 +265,8 @@ interface WithCreate extends
                 EventHubNamespace.DefinitionStages.WithSku,
                 EventHubNamespace.DefinitionStages.WithEventHub,
                 EventHubNamespace.DefinitionStages.WithAuthorizationRule,
-                EventHubNamespace.DefinitionStages.WithThroughputConfiguration {
+                EventHubNamespace.DefinitionStages.WithThroughputConfiguration,
+                EventHubNamespace.DefinitionStages.WithKafka {
         }
     }
 

azure-mgmt-eventhub/src/main/java/com/microsoft/azure/management/eventhub/implementation/EHNamespaceInner.java

@@ -25,6 +25,12 @@ public class EHNamespaceInner extends Resource {
     @JsonProperty(value = "sku")
     private Sku sku;
 
+    /**
+     * Properties of kafka resource.
+     */
+    @JsonProperty(value = "kafkaEnabled")
+    private boolean kafkaEnabled;
+
     /**
      * Provisioning state of the Namespace.
      */
@@ -175,4 +181,23 @@ public EHNamespaceInner withMaximumThroughputUnits(Integer maximumThroughputUnit
         return this;
     }
 
+    /**
+     * Get the kafkaEnabled value.
+     *
+     * @return the kafkaEnabled value
+     */
+    public boolean kafkaEnabled() {
+        return this.kafkaEnabled;
+    }
+
+    /**
+     * Set the kafkaEnabled value.
+     *
+     * @param kafkaEnabled the kafkaEnabled value to set
+     * @return the EHNamespaceInner object itself.
+     */
+    public EHNamespaceInner withKafkaEnabled(boolean kafkaEnabled) {
+        this.kafkaEnabled = kafkaEnabled;
+        return this;
+    }
 }

azure-mgmt-eventhub/src/main/java/com/microsoft/azure/management/eventhub/implementation/EventHubNamespaceImpl.java

@@ -71,6 +71,17 @@ public boolean isAutoScaleEnabled() {
         return Utils.toPrimitiveBoolean(this.inner().isAutoInflateEnabled());
     }
 
+    @Override
+    public boolean isKafkaEnabled() {
+        return this.inner().kafkaEnabled();
+    }
+
+    @Override
+    public EventHubNamespaceImpl withKafka() {
+        this.inner().withKafkaEnabled(true);
+        return this;
+    }
+
     @Override
     public int currentThroughputUnits() {
         return Utils.toPrimitiveInt(this.inner().sku().capacity());

azure-mgmt-eventhub/src/test/java/com/microsoft/azure/management/eventhub/EventHubTests.java

@@ -166,18 +166,45 @@ public void canManageEventHubNamespaceEventHubs() throws Exception {
     }
 
     @Test
-    public void canManageEventHubNamespaceAuthorizationRules() throws Exception {
+    public void canManageEventHubNamespaceWithKafka() throws Exception {
         RG_NAME = generateRandomResourceName("javacsmrg", 15);
         final String namespaceName = SdkContext.randomResourceName("ns", 14);
 
         EventHubNamespace namespace = eventHubManager.namespaces()
                 .define(namespaceName)
                     .withRegion(REGION)
                     .withNewResourceGroup(RG_NAME)
-                    .withNewManageRule("mngRule1")
-                    .withNewSendRule("sndRule1")
+                    .withKafka()
                     .create();
 
+        Assert.assertNotNull(namespace);
+        Assert.assertNotNull(namespace.inner());
+        Assert.assertTrue(namespace.isKafkaEnabled());
+
+        EventHubNamespace namespace2 = eventHubManager.namespaces()
+                .define(namespaceName)
+                .withRegion(REGION)
+                .withNewResourceGroup(RG_NAME)
+                .create();
+
+        Assert.assertNotNull(namespace);
+        Assert.assertNotNull(namespace.inner());
+        Assert.assertFalse(namespace.isKafkaEnabled());
+    }
+
+    @Test
+    public void canManageEventHubNamespaceAuthorizationRules() throws Exception {
+        RG_NAME = generateRandomResourceName("javacsmrg", 15);
+        final String namespaceName = SdkContext.randomResourceName("ns", 14);
+
+        EventHubNamespace namespace = eventHubManager.namespaces()
+                .define(namespaceName)
+                .withRegion(REGION)
+                .withNewResourceGroup(RG_NAME)
+                .withNewManageRule("mngRule1")
+                .withNewSendRule("sndRule1")
+                .create();
+
         Assert.assertNotNull(namespace);
         Assert.assertNotNull(namespace.inner());
 
@@ -202,10 +229,10 @@ public void canManageEventHubNamespaceAuthorizationRules() throws Exception {
 
         eventHubManager.namespaces()
                 .authorizationRules()
-                    .define("sndRule2")
-                    .withExistingNamespaceId(namespace.id())
-                    .withSendAccess()
-                    .create();
+                .define("sndRule2")
+                .withExistingNamespaceId(namespace.id())
+                .withSendAccess()
+                .create();
 
         rules = namespace.listAuthorizationRules();
         set.clear();