From a055e2dd57ec8afbc9b658fafc156ef7d1e51e52 Mon Sep 17 00:00:00 2001
From: twthorn <thomaswilliamthornton@gmail.com>
Date: Fri, 12 Jul 2024 17:14:30 -0400
Subject: [PATCH] DBZ-8053 Use configured shards for get tables

---
 .../connector/vitess/VitessMetadata.java      | 25 ++++++-------------
 .../connector/vitess/VitessMetadataTest.java  |  5 ----
 2 files changed, 8 insertions(+), 22 deletions(-)

diff --git a/src/main/java/io/debezium/connector/vitess/VitessMetadata.java b/src/main/java/io/debezium/connector/vitess/VitessMetadata.java
index 61f23021..dde2aa32 100644
--- a/src/main/java/io/debezium/connector/vitess/VitessMetadata.java
+++ b/src/main/java/io/debezium/connector/vitess/VitessMetadata.java
@@ -35,7 +35,7 @@ public class VitessMetadata {
     public static List<String> getShards(VitessConnectorConfig config) {
         List<String> shards;
         if (config.excludeEmptyShards()) {
-            LOGGER.info("Excluding empty shards from shard list");
+            LOGGER.info("Excluding empty shards");
             shards = getVitessShardsFromTablets(config);
         }
         else {
@@ -51,18 +51,17 @@ public static List<String> getTables(VitessConnectorConfig config) {
         if (config.excludeEmptyShards()) {
             query = "SHOW TABLES";
             List<String> shardsToQuery;
-            List<String> nonEmptyShards = getVitessShardsFromTablets(config);
-            // If there is a shard list specified, then query one of its non-empty shards
             if (config.getShard() != null && !config.getShard().isEmpty()) {
-                List<String> shardList = config.getShard();
-                shardsToQuery = intersect(shardList, nonEmptyShards);
+                LOGGER.info("Getting tables from one of the configured shards");
+                shardsToQuery = config.getShard();
             }
             else {
-                shardsToQuery = nonEmptyShards;
+                LOGGER.info("Getting tables from a non-empty shard");
+                shardsToQuery = getVitessShardsFromTablets(config);
             }
-            String randomNonEmptyShard = shardsToQuery.get(new Random().nextInt(shardsToQuery.size()));
-            LOGGER.info("Get table list from non-empty shard: {}", randomNonEmptyShard);
-            response = executeQuery(config, query, randomNonEmptyShard);
+            String randomShard = shardsToQuery.get(new Random().nextInt(shardsToQuery.size()));
+            LOGGER.info("Get tables from shard: {}", randomShard);
+            response = executeQuery(config, query, randomShard);
         }
         else {
             query = String.format("SHOW TABLES FROM %s", config.getKeyspace());
@@ -194,12 +193,4 @@ protected static List<String> flattenAndConcat(List<List<String>> nestedList) {
                 .map(innerList -> String.join("", innerList))
                 .collect(Collectors.toList());
     }
-
-    @VisibleForTesting
-    protected static List<String> intersect(List<String> list1, List<String> list2) {
-        List<String> intersection = new ArrayList<>(list1);
-        intersection.retainAll(list2);
-        return intersection;
-    }
-
 }
diff --git a/src/test/java/io/debezium/connector/vitess/VitessMetadataTest.java b/src/test/java/io/debezium/connector/vitess/VitessMetadataTest.java
index 7e6d1ba1..e04ed4d1 100644
--- a/src/test/java/io/debezium/connector/vitess/VitessMetadataTest.java
+++ b/src/test/java/io/debezium/connector/vitess/VitessMetadataTest.java
@@ -52,9 +52,4 @@ public void shouldFlattenAndConcat() {
         List<List<String>> input = List.of(List.of("foo"), List.of("bar"));
         assertThat(VitessMetadata.flattenAndConcat(input)).isEqualTo(expected);
     }
-
-    @Test
-    public void shouldIntersect() {
-        assertThat(VitessMetadata.intersect(List.of("foo", "baz"), List.of("foo", "bar"))).isEqualTo(List.of("foo"));
-    }
 }