Use the original way of validating Bigtable Change Stream table exist…

…ance to improve compatbility with testing (#31597)

Change-Id: Iba1ee679fed4276a271f2fc83c7ad657a0da8174

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIO.java

@@ -57,6 +57,7 @@
 import org.apache.beam.sdk.io.gcp.bigtable.changestreams.ChangeStreamMetrics;
 import org.apache.beam.sdk.io.gcp.bigtable.changestreams.UniqueIdGenerator;
 import org.apache.beam.sdk.io.gcp.bigtable.changestreams.action.ActionFactory;
+import org.apache.beam.sdk.io.gcp.bigtable.changestreams.dao.BigtableChangeStreamAccessor;
 import org.apache.beam.sdk.io.gcp.bigtable.changestreams.dao.BigtableClientOverride;
 import org.apache.beam.sdk.io.gcp.bigtable.changestreams.dao.DaoFactory;
 import org.apache.beam.sdk.io.gcp.bigtable.changestreams.dao.MetadataTableAdminDao;
@@ -2307,11 +2308,11 @@ public ReadChangeStream withoutValidation() {
 
     @Override
     public void validate(PipelineOptions options) {
-      BigtableServiceFactory factory = new BigtableServiceFactory();
       if (getBigtableConfig().getValidate()) {
-        try {
+        try (BigtableChangeStreamAccessor bigtableChangeStreamAccessor =
+            BigtableChangeStreamAccessor.getOrCreate(getBigtableConfig())) {
           checkArgument(
-              factory.checkTableExists(getBigtableConfig(), options, getTableId()),
+              bigtableChangeStreamAccessor.getTableAdminClient().exists(getTableId()),
               "Change Stream table %s does not exist",
               getTableId());
         } catch (IOException e) {