Sharmavijay sb omb

benchmark-framework/pom.xml

@@ -114,6 +114,12 @@
 			<version>${project.version}</version>
 		</dependency>
 
+		<dependency>
+			<groupId>${project.groupId}</groupId>
+			<artifactId>driver-azure-servicebus</artifactId>
+			<version>${project.version}</version>
+		</dependency>
+
 		<dependency>
 			<groupId>${project.groupId}</groupId>
 			<artifactId>driver-kafka</artifactId>

benchmark-framework/src/main/resources/log4j2.xml

@@ -33,10 +33,10 @@
     </Appenders>
     <Loggers>
         <Logger name="com.azure" level="error" additivity="true">
-            <AppenderRef ref="rollingFile"/>
+            <AppenderRef ref="ConsoleAppender"/>
         </Logger>
         <Root level="info">
-            <AppenderRef ref="rollingFile"/>
+            <AppenderRef ref="ConsoleAppender"/>
         </Root>
     </Loggers>
 </Configuration>

bin/perftestsuite/dedicatedV2

@@ -31,10 +31,10 @@ fi
 JVM_MEM="${HEAP_OPTS} -XX:+UseG1GC"
 
 echo "Starting KAFKA Regression Tests"
-java -server -cp $CLASSPATH $JVM_MEM io.openmessaging.benchmark.perftestsuite.dedicatedV2.KafkaRegressionTests $*
+java -server -cp $CLASSPATH $JVM_MEM io.openmessaging.benchmark.perftestsuite.eventhubs.dedicatedV2.KafkaRegressionTests $*
 
 echo "Starting AMQP Regression Tests"
-java -server -cp $CLASSPATH $JVM_MEM io.openmessaging.benchmark.perftestsuite.dedicatedV2.AMQPRegressionTests $*
+java -server -cp $CLASSPATH $JVM_MEM io.openmessaging.benchmark.perftestsuite.eventhubs.dedicatedV2.AMQPRegressionTests $*
 
 echo "Starting Multi Topic Tests"
-java -server -cp $CLASSPATH $JVM_MEM io.openmessaging.benchmark.perftestsuite.dedicatedV2.MultiTopicRegressionTests $*
+java -server -cp $CLASSPATH $JVM_MEM io.openmessaging.benchmark.perftestsuite.eventhubs.dedicatedV2.MultiTopicRegressionTests $*

bin/perftestsuite/premium

@@ -31,10 +31,10 @@ fi
 JVM_MEM="${HEAP_OPTS} -XX:+UseG1GC"
 
 echo "Starting KAFKA Regression Tests"
-java -server -cp $CLASSPATH $JVM_MEM io.openmessaging.benchmark.perftestsuite.premium.KafkaPremiumRegressionTests $*
+java -server -cp $CLASSPATH $JVM_MEM io.openmessaging.benchmark.perftestsuite.eventhubs.premium.KafkaPremiumRegressionTests $*
 
 echo "Starting AMQP Regression Tests"
-java -server -cp $CLASSPATH $JVM_MEM io.openmessaging.benchmark.perftestsuite.premium.AMQPPremiumRegressionTests $*
+java -server -cp $CLASSPATH $JVM_MEM io.openmessaging.benchmark.perftestsuite.eventhubs.premium.AMQPPremiumRegressionTests $*
 
 echo "Starting Multi Topic Tests"
-java -server -cp $CLASSPATH $JVM_MEM io.openmessaging.benchmark.perftestsuite.premium.MultiTopicPremiumRegressionTests $*
+java -server -cp $CLASSPATH $JVM_MEM io.openmessaging.benchmark.perftestsuite.eventhubs.premium.MultiTopicPremiumRegressionTests $*

bin/perftestsuite/sbstandard

@@ -0,0 +1,34 @@
+#!/bin/bash
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+echo "Azure Service Bus- Performance Regression Suite"
+TIMESTAMP=`date +%Y-%m-%d_%H-%M-%S`
+echo $TIMESTAMP
+
+if [ -d "./lib" ]; then
+        CLASSPATH=$CLASSPATH:lib/*
+else
+    CLASSPATH=perftestsuite/target/classes:`cat perftestsuite/target/classpath.txt`
+fi
+
+if [ -z "$HEAP_OPTS" ]
+then
+    HEAP_OPTS="-Xms4G -Xmx12G"
+fi
+
+JVM_MEM="${HEAP_OPTS} -XX:+UseG1GC"
+
+echo "Starting AMQP Regression Tests"
+java -server -cp $CLASSPATH $JVM_MEM io.openmessaging.benchmark.perftestsuite.servicebus.AMQPRegressionTests $*

driver-azure-servicebus/amqp-batch-standard.yaml

@@ -0,0 +1,49 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+identifier: amqp-batch-standard-sb
+name: ServiceBus AMQP Standard BatchSend
+driverClass: io.openmessaging.benchmark.driver.servicebus.ServiceBusBenchmarkDriver
+protocol: AMQP
+product: ServiceBus
+sku: Standard
+reset: false
+
+namespaceMetadata:
+  NamespaceName: sharmavijay-sbstd-glb-nm01
+  SubscriptionId: 8f8d3e6b-1a2e-4689-abc5-fd024f8c1ef2
+  ResourceGroup: Default-ServiceBus-WestUS
+  SASKeyName: RootManageSharedAccessKey
+
+topicConfig: |
+  topic.name.prefix=test-topic-amqp
+  topic.name.shared=shared-topic-amqp
+
+commonConfig: |
+  unused=0
+
+producerConfig: |
+  batch.count=100
+  batch.size=50000
+
+consumerConfig: |
+  unused=0
+
+
+

driver-azure-servicebus/pom.xml

@@ -0,0 +1,106 @@
+<!--
+
+    Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements.  See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership.  The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License.  You may obtain a copy of the License at
+
+      http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied.  See the License for the
+    specific language governing permissions and limitations
+    under the License.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+	<parent>
+		<groupId>io.openmessaging.benchmark</groupId>
+		<artifactId>messaging-benchmark</artifactId>
+		<version>0.0.2-SNAPSHOT</version>
+	</parent>
+
+	<artifactId>driver-azure-servicebus</artifactId>
+
+	<dependencies>
+		<dependency>
+			<groupId>${project.groupId}</groupId>
+			<artifactId>driver-api</artifactId>
+			<version>${project.version}</version>
+		</dependency>
+		<dependency>
+			<groupId>${project.groupId}</groupId>
+			<artifactId>app-config-adapter</artifactId>
+			<version>${project.version}</version>
+		</dependency>
+		<dependency>
+			<groupId>${project.groupId}</groupId>
+			<artifactId>storage-adapter</artifactId>
+			<version>${project.version}</version>
+		</dependency>
+		<dependency>
+			<groupId>com.azure</groupId>
+			<artifactId>azure-messaging-eventhubs</artifactId>
+			<version>5.15.2</version>
+		</dependency>
+		<dependency>
+			<groupId>com.azure</groupId>
+			<artifactId>azure-messaging-eventhubs-checkpointstore-blob</artifactId>
+			<version>1.16.1</version>
+		</dependency>
+		<dependency>
+			<groupId>com.azure.resourcemanager</groupId>
+			<artifactId>azure-resourcemanager-eventhubs</artifactId>
+			<version>2.24.0</version>
+		</dependency>
+		<dependency>
+			<groupId>com.azure</groupId>
+			<artifactId>azure-identity</artifactId>
+			<version>1.8.1</version>
+		</dependency>
+		<dependency>
+			<groupId>io.projectreactor</groupId>
+			<artifactId>reactor-core</artifactId>
+			<version>3.5.0</version>
+		</dependency>
+		<dependency>
+			<groupId>${project.groupId}</groupId>
+			<artifactId>credential-adapter</artifactId>
+			<version>${project.version}</version>
+		</dependency>
+		<dependency>
+			<groupId>com.azure</groupId>
+			<artifactId>azure-messaging-servicebus</artifactId>
+			<version>7.14.3</version>
+		</dependency>
+		<dependency>
+			<groupId>com.cedarsoftware</groupId>
+			<artifactId>java-util</artifactId>
+			<version>1.8.0</version>
+		</dependency>
+		<dependency>
+			<groupId>com.microsoft.azure</groupId>
+			<artifactId>azure-eventhubs</artifactId>
+			<version>2.2.0</version>
+		</dependency>
+	</dependencies>
+	<build>
+		<plugins>
+			<plugin>
+				<groupId>org.apache.maven.plugins</groupId>
+				<artifactId>maven-compiler-plugin</artifactId>
+				<configuration>
+					<source>8</source>
+					<target>8</target>
+				</configuration>
+			</plugin>
+		</plugins>
+	</build>
+</project>

driver-azure-servicebus/src/main/java/io/openmessaging/benchmark/driver/servicebus/Config.java

@@ -0,0 +1,29 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * <p>http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * <p>Unless required by applicable law or agreed to in writing, software distributed under the
+ * License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.openmessaging.benchmark.driver.servicebus;
+
+public class Config {
+    public String identifier;
+
+    public boolean reset;
+
+    public String topicConfig;
+
+    public String commonConfig;
+
+    public String producerConfig;
+
+    public String consumerConfig;
+}

driver-azure-servicebus/src/main/java/io/openmessaging/benchmark/driver/servicebus/ServiceBusAdministrator.java

@@ -0,0 +1,49 @@
+package io.openmessaging.benchmark.driver.servicebus;
+
+import com.azure.core.management.AzureEnvironment;
+import io.openmessaging.benchmark.appconfig.adapter.ConfigProvider;
+import io.openmessaging.benchmark.appconfig.adapter.ConfigurationKey;
+import io.openmessaging.benchmark.credential.adapter.CredentialProvider;
+import io.openmessaging.benchmark.driver.NamespaceMetadata;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+
+import static io.openmessaging.benchmark.appconfig.adapter.EnvironmentName.Production;
+
+public class ServiceBusAdministrator {
+    private static final Logger log = LoggerFactory.getLogger(ServiceBusAdministrator.class);
+
+    static NamespaceMetadata metadata;
+    static ConfigProvider configProvider;
+    static CredentialProvider credentialProvider;
+    static AzureEnvironment azureEnvironment;
+
+    public ServiceBusAdministrator(NamespaceMetadata namespaceMetadata) {
+        credentialProvider = CredentialProvider.getInstance();
+        metadata = namespaceMetadata;
+        configProvider = ConfigProvider.getInstance();
+
+        if (configProvider.getEnvironmentStage().equalsIgnoreCase(Production.name())) {
+            azureEnvironment = AzureEnvironment.AZURE;
+        } else {
+            //Configure non Azure Prod Endpoint in Azure Config.
+            azureEnvironment = new AzureEnvironment(new HashMap<String, String>() {{
+                put("managementEndpointUrl", "https://management.core.windows.net/");
+                put("resourceManagerEndpointUrl", configProvider.getConfigurationValue(ConfigurationKey.ResourceManagementURL));
+                put("activeDirectoryEndpointUrl", configProvider.getConfigurationValue(ConfigurationKey.AuthorityHost));
+            }});
+        }
+    }
+
+    public void createTopic(String topic, int partitions) {
+        //No-OP
+        //Assuming that the topicName provided is already created.
+    }
+
+    public void deleteExistingEntities(String resourceGroup, String namespaceName) {
+        log.info("Deleting existing entities");
+        //NO-OP
+    }
+}

driver-azure-servicebus/src/main/java/io/openmessaging/benchmark/driver/servicebus/ServiceBusBenchmarkConsumer.java

@@ -0,0 +1,55 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * <p>http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * <p>Unless required by applicable law or agreed to in writing, software distributed under the
+ * License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.openmessaging.benchmark.driver.servicebus;
+
+import com.azure.messaging.servicebus.ServiceBusProcessorClient;
+import com.azure.messaging.servicebus.ServiceBusReceivedMessageContext;
+import io.openmessaging.benchmark.driver.BenchmarkConsumer;
+import io.openmessaging.benchmark.driver.ConsumerCallback;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+public class ServiceBusBenchmarkConsumer implements BenchmarkConsumer {
+    private static final Logger log = LoggerFactory.getLogger(ServiceBusBenchmarkConsumer.class);
+
+    private final ServiceBusProcessorClient eventProcessorClient;
+    private final ExecutorService executor;
+    private final Future<?> consumerTask;
+
+    public ServiceBusBenchmarkConsumer(ServiceBusProcessorClient eventProcessorClient) {
+        this.eventProcessorClient = eventProcessorClient;
+        this.executor = Executors.newSingleThreadExecutor();
+        this.consumerTask = this.executor.submit(eventProcessorClient::start);
+    }
+
+    public static void processEvent(ServiceBusReceivedMessageContext eventContext, ConsumerCallback consumerCallback) {
+        consumerCallback.messageReceived(eventContext.getMessage().getBody().toBytes(),
+                TimeUnit.MILLISECONDS.toNanos(Long.parseLong(eventContext.getMessage().getEnqueuedTime().toString())));
+        eventContext.complete();
+    }
+
+    @Override
+    public void close() throws Exception {
+        log.warn("Shutting down SB Consumer gracefully");
+        executor.shutdown();
+        consumerTask.get();
+        eventProcessorClient.stop();
+    }
+}