From 7fd56a00fd1239c450480fbb672f98d9f6998c3f Mon Sep 17 00:00:00 2001 From: Yi Hu Date: Mon, 30 Mar 2026 17:25:03 -0400 Subject: [PATCH] Clean up dependencies * Delete it-splunk fork per TODO * Do not attempt to shade test jar --- it/pom.xml | 1 - it/splunk/pom.xml | 50 -- .../splunk/CustomSplunkResourceManager.java | 443 ------------------ .../conditions/CustomSplunkEventsCheck.java | 104 ---- .../it/splunk/conditions/package-info.java | 20 - .../apache/beam/it/splunk/package-info.java | 20 - v1/pom.xml | 9 - .../teleport/templates/PubSubToSplunkIT.java | 16 +- 8 files changed, 6 insertions(+), 657 deletions(-) delete mode 100644 it/splunk/pom.xml delete mode 100644 it/splunk/src/main/java/org/apache/beam/it/splunk/CustomSplunkResourceManager.java delete mode 100644 it/splunk/src/main/java/org/apache/beam/it/splunk/conditions/CustomSplunkEventsCheck.java delete mode 100644 it/splunk/src/main/java/org/apache/beam/it/splunk/conditions/package-info.java delete mode 100644 it/splunk/src/main/java/org/apache/beam/it/splunk/package-info.java diff --git a/it/pom.xml b/it/pom.xml index f1313826ef..a08da7bb49 100644 --- a/it/pom.xml +++ b/it/pom.xml @@ -199,7 +199,6 @@ datadog conditions jdbc - splunk cassandra iceberg diff --git a/it/splunk/pom.xml b/it/splunk/pom.xml deleted file mode 100644 index 17af90b974..0000000000 --- a/it/splunk/pom.xml +++ /dev/null @@ -1,50 +0,0 @@ - - - - - com.google.cloud.teleport - integration-testing-lib - 1.0-SNAPSHOT - - 4.0.0 - - it-splunk - - - - org.apache.beam - beam-it-splunk - - - org.json - json - ${json.version} - - - com.google.auto.value - auto-value - ${autovalue.version} - provided - - - com.google.auto.value - auto-value-annotations - ${autovalue.version} - - - \ No newline at end of file diff --git a/it/splunk/src/main/java/org/apache/beam/it/splunk/CustomSplunkResourceManager.java b/it/splunk/src/main/java/org/apache/beam/it/splunk/CustomSplunkResourceManager.java deleted file mode 100644 index 37af45fdcb..0000000000 --- a/it/splunk/src/main/java/org/apache/beam/it/splunk/CustomSplunkResourceManager.java +++ /dev/null @@ -1,443 +0,0 @@ -/* - * 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. - */ -package org.apache.beam.it.splunk; - -import static org.apache.beam.it.splunk.SplunkResourceManagerUtils.generateHecToken; -import static org.apache.beam.it.splunk.SplunkResourceManagerUtils.generateSplunkPassword; -import static org.apache.beam.it.splunk.SplunkResourceManagerUtils.splunkEventToMap; - -import com.splunk.Job; -import com.splunk.JobEventsArgs; -import com.splunk.ResultsReader; -import com.splunk.ResultsReaderXml; -import com.splunk.ServiceArgs; -import java.io.IOException; -import java.io.UnsupportedEncodingException; -import java.time.Duration; -import java.time.OffsetDateTime; -import java.time.format.DateTimeFormatter; -import java.util.ArrayList; -import java.util.Collection; -import java.util.Collections; -import java.util.List; -import javax.annotation.Nullable; -import org.apache.beam.it.common.ResourceManager; -import org.apache.beam.it.testcontainers.TestContainerResourceManager; -import org.apache.beam.sdk.io.splunk.SplunkEvent; -import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting; -import org.apache.http.client.methods.CloseableHttpResponse; -import org.apache.http.client.methods.HttpPost; -import org.apache.http.entity.StringEntity; -import org.apache.http.impl.client.CloseableHttpClient; -import org.awaitility.Awaitility; -import org.json.JSONObject; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.testcontainers.images.builder.Transferable; -import org.testcontainers.utility.DockerImageName; - -/** - * Client for managing Splunk resources. - * - *

The class supports one Splunk server instance. - * - *

The class is thread-safe. - * - *

Note: The Splunk TestContainer will only run on M1 Mac's if the Docker version is >= 4.16.0 - * and the "Use Rosetta for x86/amd64 emulation on Apple Silicon" setting is enabled. - * - *

TODO - Remove when Beam 2.55 is released - https://github.com/apache/beam/pull/30200 - */ -public class CustomSplunkResourceManager extends TestContainerResourceManager - implements ResourceManager { - - private static final Logger LOG = LoggerFactory.getLogger(CustomSplunkResourceManager.class); - private static final String DEFAULT_SPLUNK_CONTAINER_NAME = "splunk/splunk"; - - // A list of available Splunk Docker image tags can be found at - // https://hub.docker.com/r/splunk/splunk/tags - private static final String DEFAULT_SPLUNK_CONTAINER_TAG = "8.2"; - - // 8088 is the default port that Splunk HEC is configured to listen on - private static final int DEFAULT_SPLUNK_HEC_INTERNAL_PORT = 8088; - // 8089 is the default port that Splunkd is configured to listen on - private static final int DEFAULT_SPLUNKD_INTERNAL_PORT = 8089; - - private static final String DEFAULT_SPLUNK_USERNAME = "admin"; - - private final ServiceArgs loginArgs; - private final int hecPort; - private final String hecScheme; - private final String hecToken; - - private final SplunkClientFactory clientFactory; - - @SuppressWarnings("resource") - private CustomSplunkResourceManager(Builder builder) { - this( - new SplunkClientFactory(), - new SplunkContainer( - DockerImageName.parse(builder.containerImageName) - .withTag(builder.containerImageTag)) - .withSplunkdSslDisabled(), - builder); - } - - @VisibleForTesting - @SuppressWarnings("nullness") - CustomSplunkResourceManager( - SplunkClientFactory clientFactory, SplunkContainer container, Builder builder) { - super(setup(container, builder), builder); - - String username = DEFAULT_SPLUNK_USERNAME; - if (builder.useStaticContainer && builder.username != null) { - username = builder.username; - } - hecPort = - builder.useStaticContainer ? builder.hecPort : container.getMappedPort(builder.hecPort); - int splunkdPort = - builder.useStaticContainer - ? builder.splunkdPort - : container.getMappedPort(builder.splunkdPort); - - // TODO - add support for https scheme - String splunkdScheme = "http"; - hecScheme = "http"; - - hecToken = builder.hecToken; - - // Create Splunk service login args - this.loginArgs = new ServiceArgs(); - this.loginArgs.setPort(splunkdPort); - this.loginArgs.setHost(this.getHost()); - this.loginArgs.setScheme(splunkdScheme); - this.loginArgs.setUsername(username); - this.loginArgs.setPassword(builder.password); - - // Initialize the clients - this.clientFactory = clientFactory; - } - - /** - * Helper method for injecting config information from the builder into the given SplunkContainer. - * - * @param container The SplunkContainer before config info is injected. - * @param builder The CustomSplunkResourceManager.Builder to extract config info from. - * @return The SplunkContainer with all config info injected. - */ - @SuppressWarnings("nullness") - private static SplunkContainer setup(SplunkContainer container, Builder builder) { - // Validate builder args - if (builder.useStaticContainer) { - if (builder.hecPort < 0 || builder.splunkdPort < 0) { - throw new SplunkResourceManagerException( - "This manager was configured to use a static resource, but the hecPort and splunkdPort were not properly set."); - } - } - - builder.hecPort = builder.hecPort < 0 ? DEFAULT_SPLUNK_HEC_INTERNAL_PORT : builder.hecPort; - builder.splunkdPort = - builder.splunkdPort < 0 ? DEFAULT_SPLUNKD_INTERNAL_PORT : builder.splunkdPort; - builder.hecToken = builder.hecToken == null ? generateHecToken() : builder.hecToken; - builder.password = builder.password == null ? generateSplunkPassword() : builder.password; - // TODO - add support for ssl - return container - .withDefaultsFile( - Transferable.of( - String.format( - "splunk:%n" - + " hec:%n" - + " enable: true%n" - + " ssl: %b%n" - + " port: %s%n" - + " token: %s", - false, builder.hecPort, builder.hecToken))) - .withPassword(builder.password); - } - - public static Builder builder(String testId) { - return new Builder(testId); - } - - /** - * Returns the HTTP endpoint that this Splunk server is configured to listen on. - * - * @return the HTTP endpoint. - */ - public String getHttpEndpoint() { - return String.format("%s://%s:%d", hecScheme, getHost(), hecPort); - } - - /** - * Returns the HTTP Event Collector (HEC) endpoint that this Splunk server is configured to - * receive events at. - * - *

This will be the HTTP endpoint concatenated with '/services/collector/event'. - * - * @return the HEC service endpoint. - */ - public String getHecEndpoint() { - return getHttpEndpoint() + "/services/collector/event"; - } - - /** - * Returns the Splunk Http Event Collector (HEC) authentication token used to connect to this - * Splunk instance's HEC service. - * - * @return the HEC authentication token. - */ - public String getHecToken() { - return hecToken; - } - - /** - * Helper method for converting the given SplunkEvent into JSON format. - * - * @param event The SplunkEvent to parse. - * @return JSON String. - */ - private static String splunkEventToJson(SplunkEvent event) { - return new JSONObject(splunkEventToMap(event)).toString(); - } - - /** - * Sends the given HTTP event to the Splunk Http Event Collector (HEC) service. - * - *

Note: Setting the index field in the Splunk event requires the index already - * being configured in the Splunk instance. Unless using a static Splunk instance, omit this field - * from the event. - * - * @param event The SpunkEvent to send to the HEC service. - * @return True, if the request was successful. - */ - public synchronized boolean sendHttpEvent(SplunkEvent event) { - return sendHttpEvents(Collections.singletonList(event)); - } - - /** - * Sends the given HTTP events to the Splunk Http Event Collector (HEC) service. - * - *

Note: Setting the index field in the Splunk event requires the index already - * being configured in the Splunk instance. Unless using a static Splunk instance, omit this field - * from the events. - * - * @param events The SpunkEvents to send to the HEC service. - * @return True, if the request was successful. - */ - public synchronized boolean sendHttpEvents(Collection events) { - - LOG.info("Attempting to send {} events to {}.", events.size(), getHecEndpoint()); - - // Construct base HEC request - HttpPost httppost = new HttpPost(getHecEndpoint()); - httppost.addHeader("Authorization", "Splunk " + hecToken); - - // Loop over events and send one-by-one - StringBuilder eventsData = new StringBuilder(); - events.forEach( - event -> { - String eventStr = splunkEventToJson(event); - eventsData.append(eventStr); - LOG.info("Sending HTTP event: {}", eventStr); - }); - - try (CloseableHttpClient httpClient = clientFactory.getHttpClient()) { - // Set request data - try { - httppost.setEntity(new StringEntity(eventsData.toString())); - } catch (UnsupportedEncodingException e) { - throw new SplunkResourceManagerException( - "Error setting HTTP message data to " + eventsData, e); - } - - // Send request - try (CloseableHttpResponse response = httpClient.execute(httppost)) { - // Check error code - int code = response.getStatusLine().getStatusCode(); - if (code != 200) { - throw new SplunkResourceManagerException( - "Received http error code " + code + " sending event."); - } - } catch (Exception e) { - throw new SplunkResourceManagerException("Error sending event.", e); - } - } catch (IOException e) { - throw new SplunkResourceManagerException("Error with HTTP client.", e); - } - - LOG.info("Successfully sent {} events.", events.size()); - - return true; - } - - /** - * Return a list of all Splunk events retrieved from the Splunk server. - * - * @return All Splunk events on the server. - */ - public synchronized List getEvents() { - return getEvents("search"); - } - - /** - * Return a list of Splunk events retrieved from the Splunk server based on the given query. - * - *

e.g. query: 'search source=mySource sourcetype=mySourceType host=myHost' - * - * @param query The query to filter events by. - * @return All Splunk events on the server that match the given query. - */ - public synchronized List getEvents(String query) { - LOG.info("Reading events from Splunk using query: {}.", query); - - // Run a simple search by first creating the search job - Job job = clientFactory.getServiceClient(loginArgs).getJobs().create(query); - - // Wait up to 1 minute for search results to be ready - Awaitility.await("Retrieving events from Splunk") - .atMost(Duration.ofMinutes(1)) - .pollInterval(Duration.ofMillis(500)) - .until(job::isDone); - - // Read results - List results = new ArrayList<>(); - try { - JobEventsArgs jobEventsArgs = new JobEventsArgs(); - jobEventsArgs.setCount(0); - ResultsReader reader = new ResultsReaderXml(job.getEvents(jobEventsArgs)); - reader.forEach( - event -> - results.add( - SplunkEvent.newBuilder() - .withEvent(event.get("_raw")) - .withSource(event.get("source")) - .withSourceType(event.get("_sourcetype")) - .withHost(event.get("host")) - .withTime( - OffsetDateTime.parse( - event.get("_time"), DateTimeFormatter.ISO_OFFSET_DATE_TIME) - .toInstant() - .toEpochMilli()) - .withIndex(event.get("index")) - .create())); - - } catch (Exception e) { - throw new SplunkResourceManagerException("Error parsing XML results from Splunk.", e); - } - - LOG.info("Successfully retrieved {} events.", results.size()); - return results; - } - - /** Builder for {@link CustomSplunkResourceManager}. */ - public static final class Builder - extends TestContainerResourceManager.Builder { - - private @Nullable String username; - private @Nullable String password; - private @Nullable String hecToken; - private int hecPort; - private int splunkdPort; - - private Builder(String testId) { - super(testId, DEFAULT_SPLUNK_CONTAINER_NAME, DEFAULT_SPLUNK_CONTAINER_TAG); - this.username = null; - this.password = null; - this.hecToken = null; - this.hecPort = -1; - this.splunkdPort = -1; - } - - /** - * Set the username used to connect to a static Splunk instance. - * - *

Note: This method should only be used if {@code useStaticContainer()} is also called. - * - * @param username the username for the Splunk instance. - * @return this builder with the username manually set. - */ - public Builder setUsername(String username) { - this.username = username; - return this; - } - - /** - * Manually set the Splunk password to the given password. This password will be used by the - * resource manager to authenticate with Splunk. - * - * @param password the password for the Splunk instance. - * @return this builder with the password manually set. - */ - public Builder setPassword(String password) { - this.password = password; - return this; - } - - /** - * Manually set the Splunk HTTP Event Collector (HEC) token to the given token. This token will - * be used by the resource manager to authenticate with Splunk. - * - * @param hecToken the HEC token for the Splunk instance. - * @return this builder with the HEC token manually set. - */ - public Builder setHecToken(String hecToken) { - this.hecToken = hecToken; - return this; - } - - @Override - public Builder setHost(String containerHost) { - super.setHost(containerHost); - this.port = 0; - return this; - } - - @Override - public Builder setPort(int port) { - throw new UnsupportedOperationException( - "Please use setHecPort() and setSplunkdPort() instead."); - } - - /** - * Sets the port that the Splunk Http Event Collector (HEC) service is hosted on. - * - * @param port the port hosting the HEC service. - * @return this builder object with the HEC port set. - */ - public Builder setHecPort(int port) { - this.hecPort = port; - return this; - } - - /** - * Sets the port that the Splunkd service is hosted on. - * - * @param port the port hosting the Splunkd service. - * @return this builder object with the Splunkd port set. - */ - public Builder setSplunkdPort(int port) { - this.splunkdPort = port; - return this; - } - - @Override - public CustomSplunkResourceManager build() { - return new CustomSplunkResourceManager(this); - } - } -} diff --git a/it/splunk/src/main/java/org/apache/beam/it/splunk/conditions/CustomSplunkEventsCheck.java b/it/splunk/src/main/java/org/apache/beam/it/splunk/conditions/CustomSplunkEventsCheck.java deleted file mode 100644 index cdc42fcb2e..0000000000 --- a/it/splunk/src/main/java/org/apache/beam/it/splunk/conditions/CustomSplunkEventsCheck.java +++ /dev/null @@ -1,104 +0,0 @@ -/* - * 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. - */ -package org.apache.beam.it.splunk.conditions; - -import com.google.auto.value.AutoValue; -import javax.annotation.Nullable; -import org.apache.beam.it.conditions.ConditionCheck; -import org.apache.beam.it.splunk.CustomSplunkResourceManager; - -/** - * ConditionCheck to validate if Splunk has received a certain amount of events. - * - *

TODO - Remove when Beam 2.55 is released - https://github.com/apache/beam/pull/30200 - */ -@AutoValue -public abstract class CustomSplunkEventsCheck extends ConditionCheck { - - abstract CustomSplunkResourceManager resourceManager(); - - @Nullable - abstract String query(); - - abstract Integer minEvents(); - - @Nullable - abstract Integer maxEvents(); - - @Override - public String getDescription() { - if (maxEvents() != null) { - return String.format( - "Splunk check if logs have between %d and %d events", minEvents(), maxEvents()); - } - return String.format("Splunk check if logs have %d events", minEvents()); - } - - @Override - @SuppressWarnings("nullness") - public CheckResult check() { - long totalEvents; - if (query() != null) { - totalEvents = resourceManager().getEvents(query()).size(); - } else { - totalEvents = resourceManager().getEvents().size(); - } - if (totalEvents < minEvents()) { - return new CheckResult( - false, String.format("Expected %d but has only %d", minEvents(), totalEvents)); - } - if (maxEvents() != null && totalEvents > maxEvents()) { - return new CheckResult( - false, String.format("Expected up to %d but found %d events", maxEvents(), totalEvents)); - } - - if (maxEvents() != null) { - return new CheckResult( - true, - String.format( - "Expected between %d and %d events and found %d", - minEvents(), maxEvents(), totalEvents)); - } - - return new CheckResult( - true, String.format("Expected at least %d events and found %d", minEvents(), totalEvents)); - } - - public static Builder builder(CustomSplunkResourceManager resourceManager) { - return new AutoValue_CustomSplunkEventsCheck.Builder().setResourceManager(resourceManager); - } - - /** Builder for {@link CustomSplunkEventsCheck}. */ - @AutoValue.Builder - public abstract static class Builder { - - public abstract Builder setResourceManager(CustomSplunkResourceManager resourceManager); - - public abstract Builder setQuery(String query); - - public abstract Builder setMinEvents(Integer minEvents); - - public abstract Builder setMaxEvents(Integer maxEvents); - - abstract CustomSplunkEventsCheck autoBuild(); - - public CustomSplunkEventsCheck build() { - return autoBuild(); - } - } -} diff --git a/it/splunk/src/main/java/org/apache/beam/it/splunk/conditions/package-info.java b/it/splunk/src/main/java/org/apache/beam/it/splunk/conditions/package-info.java deleted file mode 100644 index 480e62ae43..0000000000 --- a/it/splunk/src/main/java/org/apache/beam/it/splunk/conditions/package-info.java +++ /dev/null @@ -1,20 +0,0 @@ -/* - * 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. - */ - -/** Package for managing Dataflow jobs from integration tests. */ -package org.apache.beam.it.splunk.conditions; diff --git a/it/splunk/src/main/java/org/apache/beam/it/splunk/package-info.java b/it/splunk/src/main/java/org/apache/beam/it/splunk/package-info.java deleted file mode 100644 index 4df9a97176..0000000000 --- a/it/splunk/src/main/java/org/apache/beam/it/splunk/package-info.java +++ /dev/null @@ -1,20 +0,0 @@ -/* - * 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. - */ - -/** Package for managing Dataflow jobs from integration tests. */ -package org.apache.beam.it.splunk; diff --git a/v1/pom.xml b/v1/pom.xml index 8e8ab4c241..f1465bbfd2 100644 --- a/v1/pom.xml +++ b/v1/pom.xml @@ -253,14 +253,6 @@ 3.2.3 - - - com.google.cloud.teleport - it-splunk - ${project.version} - test - - com.google.cloud.teleport @@ -992,7 +984,6 @@ false - true org.conscrypt:conscrypt-openjdk-uber diff --git a/v1/src/test/java/com/google/cloud/teleport/templates/PubSubToSplunkIT.java b/v1/src/test/java/com/google/cloud/teleport/templates/PubSubToSplunkIT.java index 30d97bb9cb..f7f12fd005 100644 --- a/v1/src/test/java/com/google/cloud/teleport/templates/PubSubToSplunkIT.java +++ b/v1/src/test/java/com/google/cloud/teleport/templates/PubSubToSplunkIT.java @@ -43,8 +43,8 @@ import org.apache.beam.it.gcp.kms.KMSResourceManager; import org.apache.beam.it.gcp.pubsub.PubsubResourceManager; import org.apache.beam.it.gcp.pubsub.conditions.PubsubMessagesCheck; -import org.apache.beam.it.splunk.CustomSplunkResourceManager; -import org.apache.beam.it.splunk.conditions.CustomSplunkEventsCheck; +import org.apache.beam.it.splunk.SplunkResourceManager; +import org.apache.beam.it.splunk.conditions.SplunkEventsCheck; import org.apache.beam.sdk.io.splunk.SplunkEvent; import org.apache.commons.lang3.RandomStringUtils; import org.json.JSONObject; @@ -55,11 +55,7 @@ import org.junit.runner.RunWith; import org.junit.runners.JUnit4; -/** - * Integration test for {@link PubSubToSplunk} classic template. - * - *

TODO - Change CustomSplunkResourceManager back when Beam 2.55 is released - */ +/** Integration test for {@link PubSubToSplunk} classic template. */ @Category({TemplateIntegrationTest.class, SkipRunnerV2Test.class}) @TemplateIntegrationTest(PubSubToSplunk.class) @RunWith(JUnit4.class) @@ -69,7 +65,7 @@ public class PubSubToSplunkIT extends TemplateTestBase { private static final int BAD_MESSAGES_COUNT = 50; private PubsubResourceManager pubsubResourceManager; - private CustomSplunkResourceManager splunkResourceManager; + private SplunkResourceManager splunkResourceManager; private KMSResourceManager kmsResourceManager; private static final String KEYRING_ID = "PubSubToSplunkIT"; @@ -84,7 +80,7 @@ public class PubSubToSplunkIT extends TemplateTestBase { public void setUp() throws IOException { pubsubResourceManager = PubsubResourceManager.builder(testName, PROJECT, credentialsProvider).build(); - splunkResourceManager = CustomSplunkResourceManager.builder(testName).build(); + splunkResourceManager = SplunkResourceManager.builder(testName).build(); kmsResourceManager = KMSResourceManager.builder(PROJECT, credentialsProvider).build(); gcsClient.createArtifact( @@ -164,7 +160,7 @@ private void testPubSubToSplunkMain( pipelineOperator() .waitForConditionAndCancel( createConfig(info), - CustomSplunkEventsCheck.builder(splunkResourceManager) + SplunkEventsCheck.builder(splunkResourceManager) .setQuery(query) .setMinEvents(allDlq ? 0 : MESSAGES_COUNT) .build(),