From 8d6297c2b1b3279af386e7d855e9df110c8cb894 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Strong=20Liu=28=E5=88=98=E5=B0=91=E5=A3=AE=29?=
 <shaozliu@microsoft.com>
Date: Sun, 26 Sep 2021 18:00:04 +0800
Subject: [PATCH] prefer api from project-reactor instead of creating our own
 wheel

---
 .../DefaultEventHubProvisioner.java           |  8 +--
 .../DefaultServiceBusQueueProvisioner.java    |  4 +-
 .../DefaultServiceBusTopicProvisioner.java    |  7 +--
 sdk/spring/azure-spring-cloud-core/pom.xml    |  6 ++
 .../com/azure/spring/core/util/Memoizer.java  | 10 +--
 .../com/azure/spring/core/util/Triple.java    | 61 -------------------
 .../com/azure/spring/core/util/Tuple.java     | 52 ----------------
 .../cloud/context/core/MemoizerTest.java      |  4 +-
 .../crud/EventHubConsumerGroupCrud.java       | 25 ++++----
 .../implementation/crud/EventHubCrud.java     | 20 +++---
 .../crud/ServiceBusQueueCrud.java             | 20 +++---
 .../crud/ServiceBusTopicCrud.java             | 20 +++---
 .../crud/ServiceBusTopicSubscriptionCrud.java | 22 +++----
 .../core/DefaultEventHubClientFactory.java    | 13 ++--
 .../eventhubs/core/EventHubRxTemplate.java    |  7 ++-
 .../eventhubs/core/EventHubTemplate.java      |  9 +--
 .../support/RxEventHubTestOperation.java      |  7 ++-
 .../DefaultServiceBusTopicClientFactory.java  | 11 ++--
 .../core/topic/ServiceBusTopicTemplate.java   |  9 +--
 19 files changed, 108 insertions(+), 207 deletions(-)
 delete mode 100644 sdk/spring/azure-spring-cloud-core/src/main/java/com/azure/spring/core/util/Triple.java
 delete mode 100644 sdk/spring/azure-spring-cloud-core/src/main/java/com/azure/spring/core/util/Tuple.java

diff --git a/sdk/spring/azure-spring-cloud-autoconfigure/src/main/java/com/azure/spring/cloud/autoconfigure/eventhub/resourcemanager/DefaultEventHubProvisioner.java b/sdk/spring/azure-spring-cloud-autoconfigure/src/main/java/com/azure/spring/cloud/autoconfigure/eventhub/resourcemanager/DefaultEventHubProvisioner.java
index e1ec79bc031de..284786cf70622 100644
--- a/sdk/spring/azure-spring-cloud-autoconfigure/src/main/java/com/azure/spring/cloud/autoconfigure/eventhub/resourcemanager/DefaultEventHubProvisioner.java
+++ b/sdk/spring/azure-spring-cloud-autoconfigure/src/main/java/com/azure/spring/cloud/autoconfigure/eventhub/resourcemanager/DefaultEventHubProvisioner.java
@@ -8,9 +8,9 @@
 import com.azure.spring.cloud.resourcemanager.implementation.crud.EventHubCrud;
 import com.azure.spring.cloud.resourcemanager.implementation.crud.EventHubNamespaceCrud;
 import com.azure.spring.core.properties.resource.AzureResourceMetadata;
-import com.azure.spring.core.util.Triple;
-import com.azure.spring.core.util.Tuple;
 import com.azure.spring.eventhubs.core.EventHubProvisioner;
+import reactor.util.function.Tuple2;
+import reactor.util.function.Tuples;
 
 /**
  * Default implementation to provision an Event Hub.
@@ -35,12 +35,12 @@ public void provisionNamespace(String namespace) {
 
     @Override
     public void provisionEventHub(String namespace, String eventHub) {
-        this.eventHubCrud.getOrCreate(Tuple.of(namespace, eventHub));
+        this.eventHubCrud.getOrCreate(Tuples.of(namespace, eventHub));
     }
 
     @Override
     public void provisionConsumerGroup(String namespace, String eventHub, String consumerGroup) {
-        this.consumerGroupCrud.getOrCreate(Triple.of(namespace, eventHub, consumerGroup));
+        this.consumerGroupCrud.getOrCreate(Tuples.of(namespace, eventHub, consumerGroup));
     }
 
 }
diff --git a/sdk/spring/azure-spring-cloud-autoconfigure/src/main/java/com/azure/spring/cloud/autoconfigure/servicebus/resourcemanager/DefaultServiceBusQueueProvisioner.java b/sdk/spring/azure-spring-cloud-autoconfigure/src/main/java/com/azure/spring/cloud/autoconfigure/servicebus/resourcemanager/DefaultServiceBusQueueProvisioner.java
index 4ae0e24518d6b..d90f888ed96e9 100644
--- a/sdk/spring/azure-spring-cloud-autoconfigure/src/main/java/com/azure/spring/cloud/autoconfigure/servicebus/resourcemanager/DefaultServiceBusQueueProvisioner.java
+++ b/sdk/spring/azure-spring-cloud-autoconfigure/src/main/java/com/azure/spring/cloud/autoconfigure/servicebus/resourcemanager/DefaultServiceBusQueueProvisioner.java
@@ -6,8 +6,8 @@
 import com.azure.resourcemanager.AzureResourceManager;
 import com.azure.spring.cloud.resourcemanager.implementation.crud.ServiceBusQueueCrud;
 import com.azure.spring.core.properties.resource.AzureResourceMetadata;
-import com.azure.spring.core.util.Tuple;
 import com.azure.spring.servicebus.core.ServiceBusQueueProvisioner;
+import reactor.util.function.Tuples;
 
 /**
  * A default implementation to provision Service Bus Queue.
@@ -23,7 +23,7 @@ public DefaultServiceBusQueueProvisioner(AzureResourceManager azureResourceManag
 
     @Override
     public void provisionQueue(String namespace, String queue) {
-        this.serviceBusQueueCrud.getOrCreate(Tuple.of(namespace, queue));
+        this.serviceBusQueueCrud.getOrCreate(Tuples.of(namespace, queue));
     }
 
 }
diff --git a/sdk/spring/azure-spring-cloud-autoconfigure/src/main/java/com/azure/spring/cloud/autoconfigure/servicebus/resourcemanager/DefaultServiceBusTopicProvisioner.java b/sdk/spring/azure-spring-cloud-autoconfigure/src/main/java/com/azure/spring/cloud/autoconfigure/servicebus/resourcemanager/DefaultServiceBusTopicProvisioner.java
index 2363e7b409b20..dce1e972209c0 100644
--- a/sdk/spring/azure-spring-cloud-autoconfigure/src/main/java/com/azure/spring/cloud/autoconfigure/servicebus/resourcemanager/DefaultServiceBusTopicProvisioner.java
+++ b/sdk/spring/azure-spring-cloud-autoconfigure/src/main/java/com/azure/spring/cloud/autoconfigure/servicebus/resourcemanager/DefaultServiceBusTopicProvisioner.java
@@ -7,9 +7,8 @@
 import com.azure.spring.cloud.resourcemanager.implementation.crud.ServiceBusTopicCrud;
 import com.azure.spring.cloud.resourcemanager.implementation.crud.ServiceBusTopicSubscriptionCrud;
 import com.azure.spring.core.properties.resource.AzureResourceMetadata;
-import com.azure.spring.core.util.Triple;
-import com.azure.spring.core.util.Tuple;
 import com.azure.spring.servicebus.core.ServiceBusTopicProvisioner;
+import reactor.util.function.Tuples;
 
 /**
  * A default implementation to provision Service Bus Topic.
@@ -27,11 +26,11 @@ public DefaultServiceBusTopicProvisioner(AzureResourceManager azureResourceManag
 
     @Override
     public void provisionTopic(String namespace, String topic) {
-        this.topicCrud.getOrCreate(Tuple.of(namespace, topic));
+        this.topicCrud.getOrCreate(Tuples.of(namespace, topic));
     }
 
     @Override
     public void provisionSubscription(String namespace, String topic, String subscription) {
-        this.subscriptionCrud.getOrCreate(Triple.of(namespace, topic, subscription));
+        this.subscriptionCrud.getOrCreate(Tuples.of(namespace, topic, subscription));
     }
 }
diff --git a/sdk/spring/azure-spring-cloud-core/pom.xml b/sdk/spring/azure-spring-cloud-core/pom.xml
index 955b33ed0220b..fdcf5dc9a1ddf 100644
--- a/sdk/spring/azure-spring-cloud-core/pom.xml
+++ b/sdk/spring/azure-spring-cloud-core/pom.xml
@@ -64,6 +64,12 @@
       <version>5.7.2</version> <!-- {x-version-update;org.junit.jupiter:junit-jupiter;external_dependency} -->
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.springframework</groupId>
+      <artifactId>spring-test</artifactId>
+      <version>5.3.9</version> <!-- {x-version-update;org.springframework:spring-test;external_dependency} -->
+      <scope>test</scope>
+    </dependency>
 
     <!-- Used to eliminate warnings when generate Java Docs.
         Added this dependency to include necessary annotations used by reactor core.
diff --git a/sdk/spring/azure-spring-cloud-core/src/main/java/com/azure/spring/core/util/Memoizer.java b/sdk/spring/azure-spring-cloud-core/src/main/java/com/azure/spring/core/util/Memoizer.java
index c47e8518d7450..716e189af4712 100644
--- a/sdk/spring/azure-spring-cloud-core/src/main/java/com/azure/spring/core/util/Memoizer.java
+++ b/sdk/spring/azure-spring-cloud-core/src/main/java/com/azure/spring/core/util/Memoizer.java
@@ -7,6 +7,8 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.BiFunction;
 import java.util.function.Function;
+import reactor.util.function.Tuple2;
+import reactor.util.function.Tuples;
 
 /**
  * Memorize function execution result
@@ -25,11 +27,11 @@ public static <T, R> Function<T, R> memoize(Map<T, R> map, Function<T, R> fn) {
     }
 
     public static <T, U, R> BiFunction<T, U, R> memoize(BiFunction<T, U, R> biFunction) {
-        Map<Tuple<T, U>, R> map = new ConcurrentHashMap<>();
-        return (t, u) -> map.computeIfAbsent(Tuple.of(t, u), (k) -> biFunction.apply(k.getFirst(), k.getSecond()));
+        Map<Tuple2<T, U>, R> map = new ConcurrentHashMap<>();
+        return (t, u) -> map.computeIfAbsent(Tuples.of(t, u), (k) -> biFunction.apply(k.getT1(), k.getT2()));
     }
 
-    public static <T, U, R> BiFunction<T, U, R> memoize(Map<Tuple<T, U>, R> map, BiFunction<T, U, R> biFunction) {
-        return (t, u) -> map.computeIfAbsent(Tuple.of(t, u), (k) -> biFunction.apply(k.getFirst(), k.getSecond()));
+    public static <T, U, R> BiFunction<T, U, R> memoize(Map<Tuple2<T, U>, R> map, BiFunction<T, U, R> biFunction) {
+        return (t, u) -> map.computeIfAbsent(Tuples.of(t, u), (k) -> biFunction.apply(k.getT1(), k.getT2()));
     }
 }
diff --git a/sdk/spring/azure-spring-cloud-core/src/main/java/com/azure/spring/core/util/Triple.java b/sdk/spring/azure-spring-cloud-core/src/main/java/com/azure/spring/core/util/Triple.java
deleted file mode 100644
index da2c310d68a07..0000000000000
--- a/sdk/spring/azure-spring-cloud-core/src/main/java/com/azure/spring/core/util/Triple.java
+++ /dev/null
@@ -1,61 +0,0 @@
-// Copyright (c) Microsoft Corporation. All rights reserved.
-// Licensed under the MIT License.
-
-package com.azure.spring.core.util;
-
-import java.util.Objects;
-
-/**
- * Tuple class.
- *
- * @param <F> The type of the first element in the triple.
- * @param <S> The type of the second element in the triple.
- * @param <T> The type of the third element in the triple.
- */
-public final class Triple<F, S, T> {
-
-    private final F first;
-    private final S second;
-    private final T third;
-
-    private Triple(F first, S second, T third) {
-        this.first = first;
-        this.second = second;
-        this.third = third;
-    }
-
-    public static <F, S, T> Triple<F, S, T> of(F first, S second, T third) {
-        return new Triple<>(first, second, third);
-    }
-
-    public F getFirst() {
-        return first;
-    }
-
-    public S getSecond() {
-        return second;
-    }
-
-    public T getThird() {
-        return third;
-    }
-
-    @Override
-    public boolean equals(Object o) {
-        if (this == o) {
-            return true;
-        }
-        if (o == null || getClass() != o.getClass()) {
-            return false;
-        }
-        Triple<?, ?, ?> triple = (Triple<?, ?, ?>) o;
-        return Objects.equals(first, triple.first)
-                   && Objects.equals(second, triple.second)
-                   && Objects.equals(third, triple.third);
-    }
-
-    @Override
-    public int hashCode() {
-        return Objects.hash(first, second, third);
-    }
-}
diff --git a/sdk/spring/azure-spring-cloud-core/src/main/java/com/azure/spring/core/util/Tuple.java b/sdk/spring/azure-spring-cloud-core/src/main/java/com/azure/spring/core/util/Tuple.java
deleted file mode 100644
index 157dbd8cd4529..0000000000000
--- a/sdk/spring/azure-spring-cloud-core/src/main/java/com/azure/spring/core/util/Tuple.java
+++ /dev/null
@@ -1,52 +0,0 @@
-// Copyright (c) Microsoft Corporation. All rights reserved.
-// Licensed under the MIT License.
-
-package com.azure.spring.core.util;
-
-import java.util.Objects;
-
-/**
- * Tuple class.
- * @param <T> The type of the first element in the tuple.
- * @param <U> The type of the second element in the tuple.
- */
-public final class Tuple<T, U> {
-
-    private final T first;
-    private final U second;
-
-    private Tuple(T first, U second) {
-        this.first = first;
-        this.second = second;
-    }
-
-    public static <T, U> Tuple<T, U> of(T first, U second) {
-        return new Tuple<>(first, second);
-    }
-
-    public T getFirst() {
-        return first;
-    }
-
-    public U getSecond() {
-        return second;
-    }
-
-    @Override
-    public boolean equals(Object o) {
-        if (this == o) {
-            return true;
-        }
-        if (o == null || getClass() != o.getClass()) {
-            return false;
-        }
-        Tuple<?, ?> tuple = (Tuple<?, ?>) o;
-        return Objects.equals(first, tuple.first) && Objects.equals(second, tuple.second);
-    }
-
-    @Override
-    public int hashCode() {
-
-        return Objects.hash(first, second);
-    }
-}
diff --git a/sdk/spring/azure-spring-cloud-core/src/test/java/com/azure/spring/cloud/context/core/MemoizerTest.java b/sdk/spring/azure-spring-cloud-core/src/test/java/com/azure/spring/cloud/context/core/MemoizerTest.java
index 0fb40b945c828..318326f784b45 100644
--- a/sdk/spring/azure-spring-cloud-core/src/test/java/com/azure/spring/cloud/context/core/MemoizerTest.java
+++ b/sdk/spring/azure-spring-cloud-core/src/test/java/com/azure/spring/cloud/context/core/MemoizerTest.java
@@ -4,7 +4,6 @@
 package com.azure.spring.cloud.context.core;
 
 import com.azure.spring.core.util.Memoizer;
-import com.azure.spring.core.util.Tuple;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
@@ -12,6 +11,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.BiFunction;
 import java.util.function.Function;
+import reactor.util.function.Tuple2;
 
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
@@ -58,7 +58,7 @@ public void memoizeFuncWithMap() {
 
     @Test
     public void memoizeBiFuncWithMap() {
-        Map<Tuple<String, String>, String> map = new ConcurrentHashMap<>();
+        Map<Tuple2<String, String>, String> map = new ConcurrentHashMap<>();
         ExpensiveBiOperation expensiveOperation = mock(ExpensiveBiOperation.class);
         when(expensiveOperation.compute(INPUT, INPUT2)).thenReturn(OUTPUT);
         BiFunction<String, String, String> memoized = Memoizer.memoize(map, expensiveOperation::compute);
diff --git a/sdk/spring/azure-spring-cloud-resourcemanager/src/main/java/com/azure/spring/cloud/resourcemanager/implementation/crud/EventHubConsumerGroupCrud.java b/sdk/spring/azure-spring-cloud-resourcemanager/src/main/java/com/azure/spring/cloud/resourcemanager/implementation/crud/EventHubConsumerGroupCrud.java
index 22048e0f3a9b2..b53afc580dc8f 100644
--- a/sdk/spring/azure-spring-cloud-resourcemanager/src/main/java/com/azure/spring/cloud/resourcemanager/implementation/crud/EventHubConsumerGroupCrud.java
+++ b/sdk/spring/azure-spring-cloud-resourcemanager/src/main/java/com/azure/spring/cloud/resourcemanager/implementation/crud/EventHubConsumerGroupCrud.java
@@ -7,14 +7,15 @@
 import com.azure.resourcemanager.AzureResourceManager;
 import com.azure.resourcemanager.eventhubs.models.EventHubConsumerGroup;
 import com.azure.spring.core.properties.resource.AzureResourceMetadata;
-import com.azure.spring.core.util.Triple;
-import com.azure.spring.core.util.Tuple;
+import reactor.util.function.Tuple3;
+import reactor.util.function.Tuples;
+
 
 /**
  * Resource manager for Event Hubs consumer group.
  */
 public class EventHubConsumerGroupCrud
-    extends AbstractResourceCrud<EventHubConsumerGroup, Triple<String, String, String>> {
+    extends AbstractResourceCrud<EventHubConsumerGroup, Tuple3<String, String, String>> {
 
     public EventHubConsumerGroupCrud(AzureResourceManager azureResourceManager,
                                      AzureResourceMetadata resourceMetadata) {
@@ -22,8 +23,8 @@ public EventHubConsumerGroupCrud(AzureResourceManager azureResourceManager,
     }
 
     @Override
-    String getResourceName(Triple<String, String, String> key) {
-        return key.getThird();
+    String getResourceName(Tuple3<String, String, String> key) {
+        return key.getT3();
     }
 
     @Override
@@ -32,13 +33,13 @@ String getResourceType() {
     }
 
     @Override
-    public EventHubConsumerGroup internalGet(Triple<String, String, String> consumerGroupCoordinate) {
+    public EventHubConsumerGroup internalGet(Tuple3<String, String, String> consumerGroupCoordinate) {
         try {
             return this.resourceManager
                 .eventHubs()
                 .consumerGroups()
-                .getByName(this.resourceMetadata.getResourceGroup(), consumerGroupCoordinate.getFirst(),
-                           consumerGroupCoordinate.getSecond(), consumerGroupCoordinate.getThird());
+                .getByName(this.resourceMetadata.getResourceGroup(), consumerGroupCoordinate.getT1(),
+                           consumerGroupCoordinate.getT2(), consumerGroupCoordinate.getT3());
         } catch (ManagementException e) {
             if (e.getResponse().getStatusCode() == 404) {
                 return null;
@@ -49,14 +50,14 @@ public EventHubConsumerGroup internalGet(Triple<String, String, String> consumer
     }
 
     @Override
-    public EventHubConsumerGroup internalCreate(Triple<String, String, String> consumerGroupCoordinate) {
+    public EventHubConsumerGroup internalCreate(Tuple3<String, String, String> consumerGroupCoordinate) {
         return this.resourceManager
             .eventHubs()
             .consumerGroups()
-            .define(consumerGroupCoordinate.getSecond())
+            .define(consumerGroupCoordinate.getT2())
             .withExistingEventHub(new EventHubCrud(this.resourceManager, this.resourceMetadata)
-                                      .getOrCreate(Tuple.of(consumerGroupCoordinate.getFirst(),
-                                                            consumerGroupCoordinate.getSecond())))
+                                      .getOrCreate(Tuples.of(consumerGroupCoordinate.getT1(),
+                                                            consumerGroupCoordinate.getT2())))
             .create();
     }
 }
diff --git a/sdk/spring/azure-spring-cloud-resourcemanager/src/main/java/com/azure/spring/cloud/resourcemanager/implementation/crud/EventHubCrud.java b/sdk/spring/azure-spring-cloud-resourcemanager/src/main/java/com/azure/spring/cloud/resourcemanager/implementation/crud/EventHubCrud.java
index ac35363df723a..4b84a415b2db3 100644
--- a/sdk/spring/azure-spring-cloud-resourcemanager/src/main/java/com/azure/spring/cloud/resourcemanager/implementation/crud/EventHubCrud.java
+++ b/sdk/spring/azure-spring-cloud-resourcemanager/src/main/java/com/azure/spring/cloud/resourcemanager/implementation/crud/EventHubCrud.java
@@ -7,20 +7,20 @@
 import com.azure.resourcemanager.AzureResourceManager;
 import com.azure.resourcemanager.eventhubs.models.EventHub;
 import com.azure.spring.core.properties.resource.AzureResourceMetadata;
-import com.azure.spring.core.util.Tuple;
+import reactor.util.function.Tuple2;
 
 /**
  * Resource manager for Event Hubs.
  */
-public class EventHubCrud extends AbstractResourceCrud<EventHub, Tuple<String, String>> {
+public class EventHubCrud extends AbstractResourceCrud<EventHub, Tuple2<String, String>> {
 
     public EventHubCrud(AzureResourceManager azureResourceManager, AzureResourceMetadata azureResourceMetadata) {
         super(azureResourceManager, azureResourceMetadata);
     }
 
     @Override
-    String getResourceName(Tuple<String, String> key) {
-        return key.getSecond();
+    String getResourceName(Tuple2<String, String> key) {
+        return key.getT2();
     }
 
     @Override
@@ -29,12 +29,12 @@ String getResourceType() {
     }
 
     @Override
-    public EventHub internalGet(Tuple<String, String> namespaceAndName) {
+    public EventHub internalGet(Tuple2<String, String> namespaceAndName) {
         try {
             return this.resourceManager.eventHubs()
                                        .getByName(this.resourceMetadata.getResourceGroup(),
-                                                  namespaceAndName.getFirst(),
-                                                  namespaceAndName.getSecond());
+                                                  namespaceAndName.getT1(),
+                                                  namespaceAndName.getT2());
         } catch (ManagementException e) {
             if (e.getResponse().getStatusCode() == 404) {
                 return null;
@@ -45,12 +45,12 @@ public EventHub internalGet(Tuple<String, String> namespaceAndName) {
     }
 
     @Override
-    public EventHub internalCreate(Tuple<String, String> namespaceAndName) {
+    public EventHub internalCreate(Tuple2<String, String> namespaceAndName) {
         return this.resourceManager.eventHubs()
-                                   .define(namespaceAndName.getSecond())
+                                   .define(namespaceAndName.getT2())
                                    .withExistingNamespace(new EventHubNamespaceCrud(this.resourceManager,
                                                                                     this.resourceMetadata)
-                                                              .getOrCreate(namespaceAndName.getFirst()))
+                                                              .getOrCreate(namespaceAndName.getT1()))
                                    .create();
     }
 }
diff --git a/sdk/spring/azure-spring-cloud-resourcemanager/src/main/java/com/azure/spring/cloud/resourcemanager/implementation/crud/ServiceBusQueueCrud.java b/sdk/spring/azure-spring-cloud-resourcemanager/src/main/java/com/azure/spring/cloud/resourcemanager/implementation/crud/ServiceBusQueueCrud.java
index c36bc980be427..afa7d099d0794 100644
--- a/sdk/spring/azure-spring-cloud-resourcemanager/src/main/java/com/azure/spring/cloud/resourcemanager/implementation/crud/ServiceBusQueueCrud.java
+++ b/sdk/spring/azure-spring-cloud-resourcemanager/src/main/java/com/azure/spring/cloud/resourcemanager/implementation/crud/ServiceBusQueueCrud.java
@@ -7,12 +7,12 @@
 import com.azure.resourcemanager.AzureResourceManager;
 import com.azure.resourcemanager.servicebus.models.Queue;
 import com.azure.spring.core.properties.resource.AzureResourceMetadata;
-import com.azure.spring.core.util.Tuple;
+import reactor.util.function.Tuple2;
 
 /**
  * Resource manager for Service Bus queue.
  */
-public class ServiceBusQueueCrud extends AbstractResourceCrud<Queue, Tuple<String, String>> {
+public class ServiceBusQueueCrud extends AbstractResourceCrud<Queue, Tuple2<String, String>> {
 
 
     public ServiceBusQueueCrud(AzureResourceManager azureResourceManager, AzureResourceMetadata azureResourceMetadata) {
@@ -20,8 +20,8 @@ public ServiceBusQueueCrud(AzureResourceManager azureResourceManager, AzureResou
     }
 
     @Override
-    String getResourceName(Tuple<String, String> key) {
-        return key.getSecond();
+    String getResourceName(Tuple2<String, String> key) {
+        return key.getT2();
     }
 
     @Override
@@ -30,12 +30,12 @@ String getResourceType() {
     }
 
     @Override
-    public Queue internalGet(Tuple<String, String> namespaceAndName) {
+    public Queue internalGet(Tuple2<String, String> namespaceAndName) {
         try {
             return new ServiceBusNamespaceCrud(this.resourceManager, this.resourceMetadata)
-                .get(namespaceAndName.getFirst())
+                .get(namespaceAndName.getT1())
                 .queues()
-                .getByName(namespaceAndName.getSecond());
+                .getByName(namespaceAndName.getT2());
         } catch (ManagementException e) {
             if (e.getResponse().getStatusCode() == 404) {
                 return null;
@@ -46,11 +46,11 @@ public Queue internalGet(Tuple<String, String> namespaceAndName) {
     }
 
     @Override
-    public Queue internalCreate(Tuple<String, String> namespaceAndName) {
+    public Queue internalCreate(Tuple2<String, String> namespaceAndName) {
         return new ServiceBusNamespaceCrud(this.resourceManager, this.resourceMetadata)
-            .getOrCreate(namespaceAndName.getFirst())
+            .getOrCreate(namespaceAndName.getT1())
             .queues()
-            .define(namespaceAndName.getSecond())
+            .define(namespaceAndName.getT2())
             .create();
     }
 }
diff --git a/sdk/spring/azure-spring-cloud-resourcemanager/src/main/java/com/azure/spring/cloud/resourcemanager/implementation/crud/ServiceBusTopicCrud.java b/sdk/spring/azure-spring-cloud-resourcemanager/src/main/java/com/azure/spring/cloud/resourcemanager/implementation/crud/ServiceBusTopicCrud.java
index 014d7405adef7..fefd6586853ae 100644
--- a/sdk/spring/azure-spring-cloud-resourcemanager/src/main/java/com/azure/spring/cloud/resourcemanager/implementation/crud/ServiceBusTopicCrud.java
+++ b/sdk/spring/azure-spring-cloud-resourcemanager/src/main/java/com/azure/spring/cloud/resourcemanager/implementation/crud/ServiceBusTopicCrud.java
@@ -7,20 +7,20 @@
 import com.azure.resourcemanager.AzureResourceManager;
 import com.azure.resourcemanager.servicebus.models.Topic;
 import com.azure.spring.core.properties.resource.AzureResourceMetadata;
-import com.azure.spring.core.util.Tuple;
+import reactor.util.function.Tuple2;
 
 /**
  * Resource manager for Service Bus topic.
  */
-public class ServiceBusTopicCrud extends AbstractResourceCrud<Topic, Tuple<String, String>> {
+public class ServiceBusTopicCrud extends AbstractResourceCrud<Topic, Tuple2<String, String>> {
 
     public ServiceBusTopicCrud(AzureResourceManager azureResourceManager, AzureResourceMetadata azureResourceMetadata) {
         super(azureResourceManager, azureResourceMetadata);
     }
 
     @Override
-    String getResourceName(Tuple<String, String> key) {
-        return key.getSecond();
+    String getResourceName(Tuple2<String, String> key) {
+        return key.getT2();
     }
 
     @Override
@@ -29,12 +29,12 @@ String getResourceType() {
     }
 
     @Override
-    public Topic internalGet(Tuple<String, String> namespaceAndName) {
+    public Topic internalGet(Tuple2<String, String> namespaceAndName) {
         try {
             return new ServiceBusNamespaceCrud(this.resourceManager, this.resourceMetadata)
-                .get(namespaceAndName.getFirst())
+                .get(namespaceAndName.getT1())
                 .topics()
-                .getByName(namespaceAndName.getSecond());
+                .getByName(namespaceAndName.getT2());
         } catch (ManagementException e) {
             if (e.getResponse().getStatusCode() == 404) {
                 return null;
@@ -45,11 +45,11 @@ public Topic internalGet(Tuple<String, String> namespaceAndName) {
     }
 
     @Override
-    public Topic internalCreate(Tuple<String, String> namespaceAndName) {
+    public Topic internalCreate(Tuple2<String, String> namespaceAndName) {
         return new ServiceBusNamespaceCrud(this.resourceManager, this.resourceMetadata)
-            .getOrCreate(namespaceAndName.getFirst())
+            .getOrCreate(namespaceAndName.getT1())
             .topics()
-            .define(namespaceAndName.getSecond())
+            .define(namespaceAndName.getT2())
             .create();
     }
 }
diff --git a/sdk/spring/azure-spring-cloud-resourcemanager/src/main/java/com/azure/spring/cloud/resourcemanager/implementation/crud/ServiceBusTopicSubscriptionCrud.java b/sdk/spring/azure-spring-cloud-resourcemanager/src/main/java/com/azure/spring/cloud/resourcemanager/implementation/crud/ServiceBusTopicSubscriptionCrud.java
index 7b945d4b5894d..5352cfaa8bfbe 100644
--- a/sdk/spring/azure-spring-cloud-resourcemanager/src/main/java/com/azure/spring/cloud/resourcemanager/implementation/crud/ServiceBusTopicSubscriptionCrud.java
+++ b/sdk/spring/azure-spring-cloud-resourcemanager/src/main/java/com/azure/spring/cloud/resourcemanager/implementation/crud/ServiceBusTopicSubscriptionCrud.java
@@ -7,14 +7,14 @@
 import com.azure.resourcemanager.AzureResourceManager;
 import com.azure.resourcemanager.servicebus.models.ServiceBusSubscription;
 import com.azure.spring.core.properties.resource.AzureResourceMetadata;
-import com.azure.spring.core.util.Triple;
-import com.azure.spring.core.util.Tuple;
+import reactor.util.function.Tuple3;
+import reactor.util.function.Tuples;
 
 /**
  * Resource manager for Service Bus topic subscription.
  */
 public class ServiceBusTopicSubscriptionCrud extends AbstractResourceCrud<ServiceBusSubscription,
-                                                                             Triple<String, String, String>> {
+    Tuple3<String, String, String>> {
 
 
     public ServiceBusTopicSubscriptionCrud(AzureResourceManager azureResourceManager,
@@ -22,8 +22,8 @@ public ServiceBusTopicSubscriptionCrud(AzureResourceManager azureResourceManager
         super(azureResourceManager, azureResourceMetadata);
     }
     @Override
-    String getResourceName(Triple<String, String, String> key) {
-        return key.getThird();
+    String getResourceName(Tuple3<String, String, String> key) {
+        return key.getT3();
     }
 
     @Override
@@ -32,12 +32,12 @@ String getResourceType() {
     }
 
     @Override
-    public ServiceBusSubscription internalGet(Triple<String, String, String> subscriptionCoordinate) {
+    public ServiceBusSubscription internalGet(Tuple3<String, String, String> subscriptionCoordinate) {
         try {
             return new ServiceBusTopicCrud(this.resourceManager, this.resourceMetadata)
-                .get(Tuple.of(subscriptionCoordinate.getFirst(), subscriptionCoordinate.getSecond()))
+                .get(Tuples.of(subscriptionCoordinate.getT1(), subscriptionCoordinate.getT2()))
                 .subscriptions()
-                .getByName(subscriptionCoordinate.getSecond());
+                .getByName(subscriptionCoordinate.getT2());
         } catch (ManagementException e) {
             if (e.getResponse().getStatusCode() == 404) {
                 return null;
@@ -48,11 +48,11 @@ public ServiceBusSubscription internalGet(Triple<String, String, String> subscri
     }
 
     @Override
-    public ServiceBusSubscription internalCreate(Triple<String, String, String> subscriptionCoordinate) {
+    public ServiceBusSubscription internalCreate(Tuple3<String, String, String> subscriptionCoordinate) {
         return new ServiceBusTopicCrud(this.resourceManager, this.resourceMetadata)
-            .getOrCreate(Tuple.of(subscriptionCoordinate.getFirst(), subscriptionCoordinate.getSecond()))
+            .getOrCreate(Tuples.of(subscriptionCoordinate.getT1(), subscriptionCoordinate.getT2()))
             .subscriptions()
-            .define(subscriptionCoordinate.getThird())
+            .define(subscriptionCoordinate.getT3())
             .create();
     }
 }
diff --git a/sdk/spring/azure-spring-eventhubs/src/main/java/com/azure/spring/eventhubs/core/DefaultEventHubClientFactory.java b/sdk/spring/azure-spring-eventhubs/src/main/java/com/azure/spring/eventhubs/core/DefaultEventHubClientFactory.java
index 3f99024b71bde..b933bf1c6e5db 100644
--- a/sdk/spring/azure-spring-eventhubs/src/main/java/com/azure/spring/eventhubs/core/DefaultEventHubClientFactory.java
+++ b/sdk/spring/azure-spring-eventhubs/src/main/java/com/azure/spring/eventhubs/core/DefaultEventHubClientFactory.java
@@ -8,7 +8,6 @@
 import com.azure.messaging.eventhubs.EventHubProducerAsyncClient;
 import com.azure.messaging.eventhubs.EventProcessorClient;
 import com.azure.spring.core.util.Memoizer;
-import com.azure.spring.core.util.Tuple;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.DisposableBean;
@@ -19,6 +18,8 @@
 import java.util.function.BiFunction;
 import java.util.function.Consumer;
 import java.util.function.Function;
+import reactor.util.function.Tuple2;
+import reactor.util.function.Tuples;
 
 import static com.azure.spring.core.ApplicationId.AZURE_SPRING_EVENT_HUB;
 import static com.azure.spring.core.ApplicationId.VERSION;
@@ -36,11 +37,11 @@ public class DefaultEventHubClientFactory implements EventHubClientFactory, Disp
 
     // Maps used for cache and clean up clients
     // (eventHubName, consumerGroup) -> consumerClient
-    private final Map<Tuple<String, String>, EventHubConsumerAsyncClient> consumerClientMap = new ConcurrentHashMap<>();
+    private final Map<Tuple2<String, String>, EventHubConsumerAsyncClient> consumerClientMap = new ConcurrentHashMap<>();
     // eventHubName -> producerClient
     private final Map<String, EventHubProducerAsyncClient> producerClientMap = new ConcurrentHashMap<>();
     // (eventHubName, consumerGroup) -> eventProcessorClient
-    private final Map<Tuple<String, String>, EventProcessorClient> processorClientMap = new ConcurrentHashMap<>();
+    private final Map<Tuple2<String, String>, EventProcessorClient> processorClientMap = new ConcurrentHashMap<>();
     // Memoized functional client creator
     private final BiFunction<String, String, EventHubConsumerAsyncClient> eventHubConsumerClientCreator =
         Memoizer.memoize(consumerClientMap, this::createEventHubClient);
@@ -117,17 +118,17 @@ public EventHubProducerAsyncClient getOrCreateProducerClient(String eventHubName
     @Override
     public EventProcessorClient createEventProcessorClient(String eventHubName, String consumerGroup,
                                                            EventHubProcessor processor) {
-        return processorClientMap.computeIfAbsent(Tuple.of(eventHubName, consumerGroup), (t) ->
+        return processorClientMap.computeIfAbsent(Tuples.of(eventHubName, consumerGroup), (t) ->
             createEventProcessorClientInternal(eventHubName, consumerGroup, processor));
     }
 
     @Override
     public Optional<EventProcessorClient> getEventProcessorClient(String eventHubName, String consumerGroup) {
-        return Optional.ofNullable(this.processorClientMap.get(Tuple.of(eventHubName, consumerGroup)));
+        return Optional.ofNullable(this.processorClientMap.get(Tuples.of(eventHubName, consumerGroup)));
     }
 
     @Override
     public EventProcessorClient removeEventProcessorClient(String eventHubName, String consumerGroup) {
-        return this.processorClientMap.remove(Tuple.of(eventHubName, consumerGroup));
+        return this.processorClientMap.remove(Tuples.of(eventHubName, consumerGroup));
     }
 }
diff --git a/sdk/spring/azure-spring-eventhubs/src/main/java/com/azure/spring/eventhubs/core/EventHubRxTemplate.java b/sdk/spring/azure-spring-eventhubs/src/main/java/com/azure/spring/eventhubs/core/EventHubRxTemplate.java
index 8eca22b1534e0..ab0c5946d1132 100644
--- a/sdk/spring/azure-spring-eventhubs/src/main/java/com/azure/spring/eventhubs/core/EventHubRxTemplate.java
+++ b/sdk/spring/azure-spring-eventhubs/src/main/java/com/azure/spring/eventhubs/core/EventHubRxTemplate.java
@@ -3,10 +3,11 @@
 
 package com.azure.spring.eventhubs.core;
 
-import com.azure.spring.core.util.Tuple;
 import com.azure.spring.messaging.PartitionSupplier;
 import org.springframework.messaging.Message;
 import reactor.core.publisher.Mono;
+import reactor.util.function.Tuple2;
+import reactor.util.function.Tuples;
 import rx.Observable;
 import rx.subscriptions.Subscriptions;
 
@@ -20,7 +21,7 @@
  */
 public class EventHubRxTemplate extends AbstractEventHubTemplate implements EventHubRxOperation {
 
-    private final ConcurrentHashMap<Tuple<String, String>, Observable<Message<?>>> subjectByNameAndGroup =
+    private final ConcurrentHashMap<Tuple2<String, String>, Observable<Message<?>>> subjectByNameAndGroup =
         new ConcurrentHashMap<>();
 
     public EventHubRxTemplate(EventHubClientFactory clientFactory) {
@@ -46,7 +47,7 @@ public <T> Observable<Void> sendRx(String destination, Message<T> message, Parti
 
     @Override
     public Observable<Message<?>> subscribe(String destination, String consumerGroup, Class<?> messagePayloadType) {
-        Tuple<String, String> nameAndConsumerGroup = Tuple.of(destination, consumerGroup);
+        Tuple2<String, String> nameAndConsumerGroup = Tuples.of(destination, consumerGroup);
 
         subjectByNameAndGroup.computeIfAbsent(nameAndConsumerGroup, k -> Observable.<Message<?>>create(subscriber -> {
             final EventHubProcessor eventHubProcessor = new EventHubProcessor(subscriber::onNext, messagePayloadType,
diff --git a/sdk/spring/azure-spring-eventhubs/src/main/java/com/azure/spring/eventhubs/core/EventHubTemplate.java b/sdk/spring/azure-spring-eventhubs/src/main/java/com/azure/spring/eventhubs/core/EventHubTemplate.java
index 4954e36312ead..b414bea96cc14 100644
--- a/sdk/spring/azure-spring-eventhubs/src/main/java/com/azure/spring/eventhubs/core/EventHubTemplate.java
+++ b/sdk/spring/azure-spring-eventhubs/src/main/java/com/azure/spring/eventhubs/core/EventHubTemplate.java
@@ -3,7 +3,6 @@
 
 package com.azure.spring.eventhubs.core;
 
-import com.azure.spring.core.util.Tuple;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.messaging.Message;
@@ -11,6 +10,8 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.function.Consumer;
+import reactor.util.function.Tuple2;
+import reactor.util.function.Tuples;
 
 /**
  * Default implementation of {@link EventHubOperation}.
@@ -22,7 +23,7 @@ public class EventHubTemplate extends AbstractEventHubTemplate implements EventH
     private static final Logger LOG = LoggerFactory.getLogger(EventHubTemplate.class);
 
     // Use this concurrent map as set since no concurrent set which has putIfAbsent
-    private final ConcurrentMap<Tuple<String, String>, Boolean> subscribedNameAndGroup = new ConcurrentHashMap<>();
+    private final ConcurrentMap<Tuple2<String, String>, Boolean> subscribedNameAndGroup = new ConcurrentHashMap<>();
 
     public EventHubTemplate(EventHubClientFactory clientFactory) {
         super(clientFactory);
@@ -32,7 +33,7 @@ public EventHubTemplate(EventHubClientFactory clientFactory) {
     @Override
     public boolean subscribe(String destination, String consumerGroup, Consumer<Message<?>> consumer,
                              Class<?> messagePayloadType) {
-        if (subscribedNameAndGroup.putIfAbsent(Tuple.of(destination, consumerGroup), true) == null) {
+        if (subscribedNameAndGroup.putIfAbsent(Tuples.of(destination, consumerGroup), true) == null) {
             this.createEventProcessorClient(destination, consumerGroup, createEventProcessor(consumer,
                 messagePayloadType));
 
@@ -46,7 +47,7 @@ public boolean subscribe(String destination, String consumerGroup, Consumer<Mess
 
     @Override
     public boolean unsubscribe(String destination, String consumerGroup) {
-        if (subscribedNameAndGroup.remove(Tuple.of(destination, consumerGroup), true)) {
+        if (subscribedNameAndGroup.remove(Tuples.of(destination, consumerGroup), true)) {
             stopEventProcessorClient(destination, consumerGroup);
             LOG.info("Consumer unsubscribed from destination '{}' with consumer group '{}'", destination,
                 consumerGroup);
diff --git a/sdk/spring/azure-spring-eventhubs/src/main/java/com/azure/spring/eventhubs/support/RxEventHubTestOperation.java b/sdk/spring/azure-spring-eventhubs/src/main/java/com/azure/spring/eventhubs/support/RxEventHubTestOperation.java
index f74f15d00b8f8..8313ecddca922 100644
--- a/sdk/spring/azure-spring-eventhubs/src/main/java/com/azure/spring/eventhubs/support/RxEventHubTestOperation.java
+++ b/sdk/spring/azure-spring-eventhubs/src/main/java/com/azure/spring/eventhubs/support/RxEventHubTestOperation.java
@@ -4,13 +4,14 @@
 package com.azure.spring.eventhubs.support;
 
 import com.azure.messaging.eventhubs.models.EventContext;
-import com.azure.spring.core.util.Tuple;
 import com.azure.spring.messaging.PartitionSupplier;
 import com.azure.spring.eventhubs.core.EventHubClientFactory;
 import com.azure.spring.eventhubs.core.EventHubRxOperation;
 import com.azure.spring.eventhubs.core.EventHubProcessor;
 import org.springframework.messaging.Message;
 import reactor.core.publisher.Mono;
+import reactor.util.function.Tuple2;
+import reactor.util.function.Tuples;
 import rx.Observable;
 import rx.subscriptions.Subscriptions;
 
@@ -21,7 +22,7 @@
  * Rx implementation for {@link EventHubTestOperation}.
  */
 public class RxEventHubTestOperation extends EventHubTestOperation implements EventHubRxOperation {
-    private final ConcurrentHashMap<Tuple<String, String>, Observable<Message<?>>> subjectByNameAndGroup =
+    private final ConcurrentHashMap<Tuple2<String, String>, Observable<Message<?>>> subjectByNameAndGroup =
         new ConcurrentHashMap<>();
 
     public RxEventHubTestOperation(EventHubClientFactory clientFactory, Supplier<EventContext> eventContextSupplier) {
@@ -46,7 +47,7 @@ public <T> Observable<Void> sendRx(String destination, Message<T> message, Parti
 
     @Override
     public Observable<Message<?>> subscribe(String destination, String consumerGroup, Class<?> messagePayloadType) {
-        Tuple<String, String> nameAndConsumerGroup = Tuple.of(destination, consumerGroup);
+        Tuple2<String, String> nameAndConsumerGroup = Tuples.of(destination, consumerGroup);
 
         subjectByNameAndGroup.computeIfAbsent(nameAndConsumerGroup, k -> Observable.<Message<?>>create(subscriber -> {
             final EventHubProcessor eventHubProcessor = createEventProcessor(subscriber::onNext, messagePayloadType);
diff --git a/sdk/spring/azure-spring-servicebus/src/main/java/com/azure/spring/servicebus/core/DefaultServiceBusTopicClientFactory.java b/sdk/spring/azure-spring-servicebus/src/main/java/com/azure/spring/servicebus/core/DefaultServiceBusTopicClientFactory.java
index b7423cccd3008..495e10f4dad9b 100644
--- a/sdk/spring/azure-spring-servicebus/src/main/java/com/azure/spring/servicebus/core/DefaultServiceBusTopicClientFactory.java
+++ b/sdk/spring/azure-spring-servicebus/src/main/java/com/azure/spring/servicebus/core/DefaultServiceBusTopicClientFactory.java
@@ -7,7 +7,6 @@
 import com.azure.messaging.servicebus.ServiceBusClientBuilder;
 import com.azure.messaging.servicebus.ServiceBusProcessorClient;
 import com.azure.messaging.servicebus.ServiceBusSenderAsyncClient;
-import com.azure.spring.core.util.Tuple;
 import com.azure.spring.servicebus.support.ServiceBusClientConfig;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -16,6 +15,8 @@
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Consumer;
+import reactor.util.function.Tuple2;
+import reactor.util.function.Tuples;
 
 /**
  * Default implementation of {@link ServiceBusTopicClientFactory}. Client will be cached to improve performance
@@ -26,7 +27,7 @@ public class DefaultServiceBusTopicClientFactory extends AbstractServiceBusSende
     implements ServiceBusTopicClientFactory, DisposableBean {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(DefaultServiceBusTopicClientFactory.class);
-    private final Map<Tuple<String, String>, ServiceBusProcessorClient> topicProcessorMap = new ConcurrentHashMap<>();
+    private final Map<Tuple2<String, String>, ServiceBusProcessorClient> topicProcessorMap = new ConcurrentHashMap<>();
     private final Map<String, ServiceBusSenderAsyncClient> topicSenderMap = new ConcurrentHashMap<>();
 
 
@@ -57,9 +58,9 @@ public ServiceBusProcessorClient getOrCreateProcessor(
         String subscription,
         ServiceBusClientConfig clientConfig,
         ServiceBusMessageProcessor messageProcessor) {
-        return this.topicProcessorMap.computeIfAbsent(Tuple.of(topic, subscription),
-                                                      t -> createProcessor(t.getFirst(),
-                                                                           t.getSecond(),
+        return this.topicProcessorMap.computeIfAbsent(Tuples.of(topic, subscription),
+                                                      t -> createProcessor(t.getT1(),
+                                                                           t.getT2(),
                                                                            clientConfig,
                                                                            messageProcessor));
     }
diff --git a/sdk/spring/azure-spring-servicebus/src/main/java/com/azure/spring/servicebus/core/topic/ServiceBusTopicTemplate.java b/sdk/spring/azure-spring-servicebus/src/main/java/com/azure/spring/servicebus/core/topic/ServiceBusTopicTemplate.java
index 2d98ad6eff59d..1a0030e729d3c 100644
--- a/sdk/spring/azure-spring-servicebus/src/main/java/com/azure/spring/servicebus/core/topic/ServiceBusTopicTemplate.java
+++ b/sdk/spring/azure-spring-servicebus/src/main/java/com/azure/spring/servicebus/core/topic/ServiceBusTopicTemplate.java
@@ -4,7 +4,6 @@
 package com.azure.spring.servicebus.core.topic;
 
 import com.azure.messaging.servicebus.ServiceBusProcessorClient;
-import com.azure.spring.core.util.Tuple;
 import com.azure.spring.servicebus.core.DefaultServiceBusMessageProcessor;
 import com.azure.spring.servicebus.support.ServiceBusClientConfig;
 import com.azure.spring.servicebus.support.ServiceBusRuntimeException;
@@ -21,6 +20,8 @@
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Consumer;
+import reactor.util.function.Tuple2;
+import reactor.util.function.Tuples;
 
 /**
  * Default implementation of {@link ServiceBusTopicOperation}.
@@ -37,7 +38,7 @@ public class ServiceBusTopicTemplate extends ServiceBusTemplate<ServiceBusTopicC
 
     private static final String MSG_SUCCESS_CHECKPOINT = "Consumer group '%s' of topic '%s' checkpointed %s in %s mode";
 
-    private final Set<Tuple<String, String>> nameAndConsumerGroups = ConcurrentHashMap.newKeySet();
+    private final Set<Tuple2<String, String>> nameAndConsumerGroups = ConcurrentHashMap.newKeySet();
 
     public ServiceBusTopicTemplate(ServiceBusTopicClientFactory clientFactory) {
         super(clientFactory);
@@ -60,7 +61,7 @@ public boolean subscribe(String destination,
                              Class<?> payloadType) {
         Assert.hasText(destination, "destination can't be null or empty");
 
-        Tuple<String, String> nameAndConsumerGroup = Tuple.of(destination, consumerGroup);
+        Tuple2<String, String> nameAndConsumerGroup = Tuples.of(destination, consumerGroup);
 
         if (this.nameAndConsumerGroups.contains(nameAndConsumerGroup)) {
             return false;
@@ -76,7 +77,7 @@ public boolean subscribe(String destination,
     public boolean unsubscribe(String destination, String consumerGroup) {
         // TODO: unregister message handler but service bus sdk unsupported
 
-        return this.nameAndConsumerGroups.remove(Tuple.of(destination, consumerGroup));
+        return this.nameAndConsumerGroups.remove(Tuples.of(destination, consumerGroup));
     }
 
     /**