Enable SpannerChangeStreamPlacementTableIT and update IntegrationTestEnv cleanup (#38052)

Author: tianz101

sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/it/IntegrationTestEnv.java

@@ -155,7 +155,12 @@ protected void after() {
               .get(TIMEOUT_MINUTES, TimeUnit.MINUTES);
         }
       } catch (Exception e) {
-        LOG.error("Failed to drop table {}. Skipping...", table, e);
+        if (isPlacementTableBasedChangeStream) {
+          // Drop placement table requires all rows deleted and garbage collected.
+          LOG.info("Failed to drop table {}. Skipping...", table, e);
+        } else {
+          LOG.error("Failed to drop table {}. Skipping...", table, e);
+        }
       }
     }
 

sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/it/SpannerChangeStreamPlacementTableIT.java

@@ -57,7 +57,6 @@
 import org.junit.Before;
 import org.junit.BeforeClass;
 import org.junit.ClassRule;
-import org.junit.Ignore;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.ExpectedException;
@@ -66,12 +65,14 @@
 
 /** End-to-end test of Cloud Spanner placement table. */
 @RunWith(JUnit4.class)
-@Ignore(
-    "TODO: enable this test class when placement table is supported by change stream in prod."
-        + "For now this test can only be exercised mannually.")
 public class SpannerChangeStreamPlacementTableIT {
 
-  @ClassRule public static final IntegrationTestEnv ENV = new IntegrationTestEnv();
+  @ClassRule
+  public static final IntegrationTestEnv ENV =
+      new IntegrationTestEnv(
+          /*isPostgres=*/ false,
+          /*isPlacementTableBasedChangeStream=*/ true,
+          /*host=*/ Optional.empty());
 
   @Rule public final transient TestPipeline pipeline = TestPipeline.create();