From 704e379c3950ab63e74fb4d2969164fe57c9a53a Mon Sep 17 00:00:00 2001 From: Adalbert Makarovych Date: Tue, 26 May 2026 22:41:29 +0300 Subject: [PATCH 1/3] feat: add Pub/Sub emulator support for source connector --- README.md | 1 + .../pubsub/kafka/common/ConnectorUtils.java | 14 +++++ .../kafka/sink/CloudPubSubSinkTask.java | 8 +-- .../source/CloudPubSubGRPCSubscriber.java | 43 ++++++++++---- .../CloudPubSubRoundRobinSubscriber.java | 5 +- .../source/CloudPubSubSourceConnector.java | 56 ++++++++++++++----- .../kafka/source/CloudPubSubSourceTask.java | 24 ++++++-- .../CloudPubSubSourceConnectorTest.java | 7 +-- .../source/CloudPubSubSourceTaskTest.java | 38 +++++++++++++ 9 files changed, 157 insertions(+), 39 deletions(-) diff --git a/README.md b/README.md index fd63f9c4..039f91eb 100644 --- a/README.md +++ b/README.md @@ -173,6 +173,7 @@ configurations: | cps.subscription | String | REQUIRED (No default) | The Pub/Sub subscription ID, e.g. "baz" for subscription "/projects/bar/subscriptions/baz". | | cps.project | String | REQUIRED (No default) | The project containing the Pub/Sub subscription, e.g. "bar" from above. | | cps.endpoint | String | "pubsub.googleapis.com:443" | The [Pub/Sub endpoint](https://cloud.google.com/pubsub/docs/reference/service_apis_overview#service_endpoints) to use. | +| cps.useEmulator | Boolean | false | When true, use the Pub/Sub emulator instead of the production service. The emulator endpoint will be determined by the PUBSUB_EMULATOR_HOST environment variable, or fallback to the cps.endpoint configuration. | | kafka.topic | String | REQUIRED (No default) | The Kafka topic which will receive messages from the Pub/Sub subscription. | | cps.maxBatchSize | Integer | 100 | The maximum number of messages per batch in a pull request to Pub/Sub. | | cps.makeOrderingKeyAttribute | Boolean | false | When true, copy the ordering key to the set of attributes set in the Kafka message. | diff --git a/src/main/java/com/google/pubsub/kafka/common/ConnectorUtils.java b/src/main/java/com/google/pubsub/kafka/common/ConnectorUtils.java index 63ad7c0c..ece4e9a2 100644 --- a/src/main/java/com/google/pubsub/kafka/common/ConnectorUtils.java +++ b/src/main/java/com/google/pubsub/kafka/common/ConnectorUtils.java @@ -58,4 +58,18 @@ public static synchronized ScheduledExecutorService getSystemExecutor() { } return SYSTEM_EXECUTOR.get(); } + + + // Resolve the endpoint. When using the emulator, prefer PUBSUB_EMULATOR_HOST and fall back to + // the configured cps.endpoint. + public static String getPubsubEndpoint(boolean useEmulator, String cpsEndpoint) { + if (useEmulator) { + String emulatorHost = System.getenv(PUBSUB_EMULATOR_HOST); + if (emulatorHost != null && !emulatorHost.isEmpty()) { + return emulatorHost; + } + } + + return cpsEndpoint; + } } diff --git a/src/main/java/com/google/pubsub/kafka/sink/CloudPubSubSinkTask.java b/src/main/java/com/google/pubsub/kafka/sink/CloudPubSubSinkTask.java index d9abd6de..d23a9729 100644 --- a/src/main/java/com/google/pubsub/kafka/sink/CloudPubSubSinkTask.java +++ b/src/main/java/com/google/pubsub/kafka/sink/CloudPubSubSinkTask.java @@ -418,11 +418,8 @@ private void createPublisher() { // Configure endpoint, credentials and channel based on whether we're using emulator or // production + String endpoint = ConnectorUtils.getPubsubEndpoint(useEmulator, cpsEndpoint); if (useEmulator) { - // For emulator: use PUBSUB_EMULATOR_HOST env var, fallback to configured cps.endpoint, then - // default - String emulatorHost = System.getenv(ConnectorUtils.PUBSUB_EMULATOR_HOST); - String endpoint = emulatorHost != null ? emulatorHost : cpsEndpoint; builder .setCredentialsProvider(com.google.api.gax.core.NoCredentialsProvider.create()) .setChannelProvider( @@ -431,8 +428,7 @@ private void createPublisher() { .setChannelConfigurator(channel -> channel.usePlaintext()) .build()); } else { - // For production: use configured credentials and endpoint - builder.setCredentialsProvider(gcpCredentialsProvider).setEndpoint(cpsEndpoint); + builder.setCredentialsProvider(gcpCredentialsProvider).setEndpoint(endpoint); } if (orderingKeySource != OrderingKeySource.NONE) { builder.setEnableMessageOrdering(true); diff --git a/src/main/java/com/google/pubsub/kafka/source/CloudPubSubGRPCSubscriber.java b/src/main/java/com/google/pubsub/kafka/source/CloudPubSubGRPCSubscriber.java index 470abcb8..b60e9b34 100644 --- a/src/main/java/com/google/pubsub/kafka/source/CloudPubSubGRPCSubscriber.java +++ b/src/main/java/com/google/pubsub/kafka/source/CloudPubSubGRPCSubscriber.java @@ -18,10 +18,13 @@ import com.google.api.core.ApiFuture; import com.google.api.core.ApiFutures; import com.google.api.gax.core.CredentialsProvider; +import com.google.api.gax.core.NoCredentialsProvider; +import com.google.api.gax.grpc.InstantiatingGrpcChannelProvider; import com.google.cloud.pubsub.v1.stub.GrpcSubscriberStub; import com.google.cloud.pubsub.v1.stub.SubscriberStubSettings; import com.google.common.util.concurrent.MoreExecutors; import com.google.protobuf.Empty; +import com.google.pubsub.kafka.common.ConnectorUtils; import com.google.pubsub.v1.AcknowledgeRequest; import com.google.pubsub.v1.ProjectSubscriptionName; import com.google.pubsub.v1.PullRequest; @@ -48,16 +51,19 @@ public class CloudPubSubGRPCSubscriber implements CloudPubSubSubscriber { private final String endpoint; private final ProjectSubscriptionName subscriptionName; private final int cpsMaxBatchSize; + private final boolean useEmulator; CloudPubSubGRPCSubscriber( CredentialsProvider gcpCredentialsProvider, String endpoint, ProjectSubscriptionName subscriptionName, - int cpsMaxBatchSize) { + int cpsMaxBatchSize, + boolean useEmulator) { this.gcpCredentialsProvider = gcpCredentialsProvider; this.endpoint = endpoint; this.subscriptionName = subscriptionName; this.cpsMaxBatchSize = cpsMaxBatchSize; + this.useEmulator = useEmulator; makeSubscriber(); } @@ -103,15 +109,32 @@ private void makeSubscriber() { subscriber.close(); } log.info("Creating subscriber."); - SubscriberStubSettings subscriberStubSettings = - SubscriberStubSettings.newBuilder() - .setTransportChannelProvider( - SubscriberStubSettings.defaultGrpcTransportProviderBuilder() - .setMaxInboundMessageSize(20 << 20) // 20MB - .build()) - .setCredentialsProvider(gcpCredentialsProvider) - .setEndpoint(endpoint) - .build(); + + // Configure endpoint, credentials and channel based on whether we're using emulator or + // production + SubscriberStubSettings subscriberStubSettings; + if (useEmulator) { + subscriberStubSettings = + SubscriberStubSettings.newBuilder() + .setCredentialsProvider(NoCredentialsProvider.create()) + .setTransportChannelProvider( + InstantiatingGrpcChannelProvider.newBuilder() + .setMaxInboundMessageSize(20 << 20) // 20MB + .setEndpoint(endpoint) + .setChannelConfigurator(channel -> channel.usePlaintext()) + .build()) + .build(); + } else { + subscriberStubSettings = + SubscriberStubSettings.newBuilder() + .setTransportChannelProvider( + SubscriberStubSettings.defaultGrpcTransportProviderBuilder() + .setMaxInboundMessageSize(20 << 20) // 20MB + .build()) + .setCredentialsProvider(gcpCredentialsProvider) + .setEndpoint(endpoint) + .build(); + } subscriber = GrpcSubscriberStub.create(subscriberStubSettings); // We change the subscriber every 25 - 35 minutes in order to avoid GOAWAY errors. nextSubscriberResetTime = diff --git a/src/main/java/com/google/pubsub/kafka/source/CloudPubSubRoundRobinSubscriber.java b/src/main/java/com/google/pubsub/kafka/source/CloudPubSubRoundRobinSubscriber.java index d6bfcd6d..52a95de3 100644 --- a/src/main/java/com/google/pubsub/kafka/source/CloudPubSubRoundRobinSubscriber.java +++ b/src/main/java/com/google/pubsub/kafka/source/CloudPubSubRoundRobinSubscriber.java @@ -38,12 +38,13 @@ public CloudPubSubRoundRobinSubscriber( CredentialsProvider gcpCredentialsProvider, String endpoint, ProjectSubscriptionName subscriptionName, - int cpsMaxBatchSize) { + int cpsMaxBatchSize, + boolean useEmulator) { subscribers = new ArrayList<>(); for (int i = 0; i < subscriberCount; ++i) { subscribers.add( new CloudPubSubGRPCSubscriber( - gcpCredentialsProvider, endpoint, subscriptionName, cpsMaxBatchSize)); + gcpCredentialsProvider, endpoint, subscriptionName, cpsMaxBatchSize, useEmulator)); } } diff --git a/src/main/java/com/google/pubsub/kafka/source/CloudPubSubSourceConnector.java b/src/main/java/com/google/pubsub/kafka/source/CloudPubSubSourceConnector.java index 09a6eb22..0b1eee72 100644 --- a/src/main/java/com/google/pubsub/kafka/source/CloudPubSubSourceConnector.java +++ b/src/main/java/com/google/pubsub/kafka/source/CloudPubSubSourceConnector.java @@ -16,6 +16,8 @@ package com.google.pubsub.kafka.source; import com.google.api.gax.core.CredentialsProvider; +import com.google.api.gax.core.NoCredentialsProvider; +import com.google.api.gax.grpc.InstantiatingGrpcChannelProvider; import com.google.cloud.pubsub.v1.stub.GrpcSubscriberStub; import com.google.cloud.pubsub.v1.stub.SubscriberStubSettings; import com.google.common.annotations.VisibleForTesting; @@ -140,10 +142,13 @@ public void start(Map props) { Map validated = config().parse(props); String cpsProject = validated.get(ConnectorUtils.CPS_PROJECT_CONFIG).toString(); String cpsSubscription = validated.get(CPS_SUBSCRIPTION_CONFIG).toString(); + String cpsEndpoint = (String) validated.get(ConnectorUtils.CPS_ENDPOINT); + boolean useEmulator = (Boolean) validated.get(ConnectorUtils.CPS_USE_EMULATOR); + String endpoint = ConnectorUtils.getPubsubEndpoint(useEmulator, cpsEndpoint); ConnectorCredentialsProvider credentialsProvider = ConnectorCredentialsProvider.fromConfig(validated); - verifySubscription(cpsProject, cpsSubscription, credentialsProvider); + verifySubscription(cpsProject, cpsSubscription, credentialsProvider, endpoint, useEmulator); this.props = props; log.info("Started the CloudPubSubSourceConnector"); } @@ -297,7 +302,13 @@ public ConfigDef config() { Type.STRING, ConnectorUtils.CPS_DEFAULT_ENDPOINT, Importance.LOW, - "The Pub/Sub endpoint to use."); + "The Pub/Sub endpoint to use.") + .define( + ConnectorUtils.CPS_USE_EMULATOR, + Type.BOOLEAN, + false, + Importance.LOW, + "When true, use the Pub/Sub emulator instead of the production service."); } /** @@ -306,24 +317,43 @@ public ConfigDef config() { */ @VisibleForTesting public void verifySubscription( - String cpsProject, String cpsSubscription, CredentialsProvider credentialsProvider) { + String cpsProject, + String cpsSubscription, + CredentialsProvider credentialsProvider, + String endpoint, + boolean useEmulator) { try { - SubscriberStubSettings subscriberStubSettings = - SubscriberStubSettings.newBuilder() - .setTransportChannelProvider( - SubscriberStubSettings.defaultGrpcTransportProviderBuilder() - .setMaxInboundMessageSize(20 << 20) // 20MB - .build()) - .setCredentialsProvider(credentialsProvider) - .build(); - GrpcSubscriberStub stub = GrpcSubscriberStub.create(subscriberStubSettings); + // Configure endpoint, credentials and channel based on whether we're using emulator or + // production + SubscriberStubSettings.Builder settingsBuilder = SubscriberStubSettings.newBuilder(); + if (useEmulator) { + settingsBuilder + .setCredentialsProvider(NoCredentialsProvider.create()) + .setTransportChannelProvider( + InstantiatingGrpcChannelProvider.newBuilder() + .setMaxInboundMessageSize(20 << 20) // 20MB + .setEndpoint(endpoint) + .setChannelConfigurator(channel -> channel.usePlaintext()) + .build()); + } else { + settingsBuilder + .setTransportChannelProvider( + SubscriberStubSettings.defaultGrpcTransportProviderBuilder() + .setMaxInboundMessageSize(20 << 20) // 20MB + .build()) + .setCredentialsProvider(credentialsProvider) + .setEndpoint(endpoint); + } + GetSubscriptionRequest request = GetSubscriptionRequest.newBuilder() .setSubscription( String.format( ConnectorUtils.CPS_SUBSCRIPTION_FORMAT, cpsProject, cpsSubscription)) .build(); - stub.getSubscriptionCallable().call(request); + try (GrpcSubscriberStub stub = GrpcSubscriberStub.create(settingsBuilder.build())) { + stub.getSubscriptionCallable().call(request); + } } catch (Exception e) { throw new ConnectException( "Error verifying the subscription " + cpsSubscription + " for project " + cpsProject, e); diff --git a/src/main/java/com/google/pubsub/kafka/source/CloudPubSubSourceTask.java b/src/main/java/com/google/pubsub/kafka/source/CloudPubSubSourceTask.java index e84224f5..c839ba7e 100644 --- a/src/main/java/com/google/pubsub/kafka/source/CloudPubSubSourceTask.java +++ b/src/main/java/com/google/pubsub/kafka/source/CloudPubSubSourceTask.java @@ -76,6 +76,7 @@ public class CloudPubSubSourceTask extends SourceTask { private CloudPubSubSubscriber subscriber; private final Set standardAttributes = new HashSet<>(); private boolean useKafkaHeaders; + private boolean useEmulator; public CloudPubSubSourceTask() {} @@ -99,6 +100,8 @@ public void start(Map props) { validatedProps.get(CloudPubSubSourceConnector.CPS_SUBSCRIPTION_CONFIG).toString()) .build(); String cpsEndpoint = (String) validatedProps.get(ConnectorUtils.CPS_ENDPOINT); + useEmulator = (Boolean) validatedProps.get(ConnectorUtils.CPS_USE_EMULATOR); + String endpoint = ConnectorUtils.getPubsubEndpoint(useEmulator, cpsEndpoint); kafkaTopic = validatedProps.get(CloudPubSubSourceConnector.KAFKA_TOPIC_CONFIG).toString(); int cpsMaxBatchSize = (Integer) validatedProps.get(CloudPubSubSourceConnector.CPS_MAX_BATCH_SIZE_CONFIG); @@ -142,7 +145,6 @@ public void start(Map props) { receiver -> { Subscriber.Builder builder = Subscriber.newBuilder(cpsSubscription, receiver) - .setCredentialsProvider(gcpCredentialsProvider) .setFlowControlSettings( FlowControlSettings.newBuilder() .setLimitExceededBehavior(LimitExceededBehavior.Block) @@ -150,8 +152,21 @@ public void start(Map props) { .setMaxOutstandingRequestBytes(streamingPullBytes) .build()) .setParallelPullCount(streamingPullParallelStreams) - .setEndpoint(cpsEndpoint) .setExecutorProvider(FixedExecutorProvider.create(getSystemExecutor())); + // Configure endpoint, credentials and channel based on whether we're using + // emulator or production + if (useEmulator) { + builder + .setCredentialsProvider( + com.google.api.gax.core.NoCredentialsProvider.create()) + .setChannelProvider( + com.google.api.gax.grpc.InstantiatingGrpcChannelProvider.newBuilder() + .setEndpoint(endpoint) + .setChannelConfigurator(channel -> channel.usePlaintext()) + .build()); + } else { + builder.setCredentialsProvider(gcpCredentialsProvider).setEndpoint(endpoint); + } if (streamingPullMaxAckDeadlineMs > 0) { builder.setMaxAckExtensionPeriod( Duration.ofMillis(streamingPullMaxAckDeadlineMs)); @@ -168,9 +183,10 @@ public void start(Map props) { new CloudPubSubRoundRobinSubscriber( NUM_CPS_SUBSCRIBERS, gcpCredentialsProvider, - cpsEndpoint, + endpoint, cpsSubscription, - cpsMaxBatchSize), + cpsMaxBatchSize, + useEmulator), runnable -> getSystemExecutor() .scheduleAtFixedRate(runnable, 100, 100, TimeUnit.MILLISECONDS)); diff --git a/src/test/java/com/google/pubsub/kafka/source/CloudPubSubSourceConnectorTest.java b/src/test/java/com/google/pubsub/kafka/source/CloudPubSubSourceConnectorTest.java index 6d31a018..2f3a814f 100644 --- a/src/test/java/com/google/pubsub/kafka/source/CloudPubSubSourceConnectorTest.java +++ b/src/test/java/com/google/pubsub/kafka/source/CloudPubSubSourceConnectorTest.java @@ -16,8 +16,7 @@ package com.google.pubsub.kafka.source; import static org.junit.Assert.assertEquals; -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.*; import static org.mockito.Mockito.doNothing; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.spy; @@ -56,7 +55,7 @@ public void setup() { public void testStartWhenSubscriptionNonexistant() { doThrow(new ConnectException("")) .when(connector) - .verifySubscription(anyString(), anyString(), any(ConnectorCredentialsProvider.class)); + .verifySubscription(anyString(), anyString(), any(ConnectorCredentialsProvider.class), anyString(), anyBoolean()); connector.start(props); } @@ -69,7 +68,7 @@ public void testStartWhenRequiredConfigMissing() { public void testTaskConfigs() { doNothing() .when(connector) - .verifySubscription(anyString(), anyString(), any(ConnectorCredentialsProvider.class)); + .verifySubscription(anyString(), anyString(), any(ConnectorCredentialsProvider.class), anyString(), anyBoolean()); connector.start(props); List> taskConfigs = connector.taskConfigs(NUM_TASKS); assertEquals(taskConfigs.size(), NUM_TASKS); diff --git a/src/test/java/com/google/pubsub/kafka/source/CloudPubSubSourceTaskTest.java b/src/test/java/com/google/pubsub/kafka/source/CloudPubSubSourceTaskTest.java index 1e1be8e6..e0a86763 100644 --- a/src/test/java/com/google/pubsub/kafka/source/CloudPubSubSourceTaskTest.java +++ b/src/test/java/com/google/pubsub/kafka/source/CloudPubSubSourceTaskTest.java @@ -18,6 +18,7 @@ import static org.junit.Assert.assertArrayEquals; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.RETURNS_DEEP_STUBS; import static org.mockito.Mockito.mock; @@ -36,6 +37,8 @@ import java.util.HashMap; import java.util.List; import java.util.Map; + +import org.apache.kafka.common.config.ConfigDef; import org.apache.kafka.connect.data.Field; import org.apache.kafka.connect.data.Schema; import org.apache.kafka.connect.data.SchemaBuilder; @@ -116,6 +119,41 @@ public void setup() { CloudPubSubSourceConnector.PartitionScheme.ROUND_ROBIN.toString()); } + /** Tests that the emulator configuration is properly defined and parsed. */ + @Test + public void testEmulatorConfiguration() { + CloudPubSubSourceConnector connector = new CloudPubSubSourceConnector(); + ConfigDef configDef = connector.config(); + + assertTrue( + "Emulator configuration should be defined", + configDef.names().contains(ConnectorUtils.CPS_USE_EMULATOR)); + + props.put(ConnectorUtils.CPS_USE_EMULATOR, "true"); + + Map parsedProps = configDef.parse(props); + assertTrue( + "Emulator should be enabled", (Boolean) parsedProps.get(ConnectorUtils.CPS_USE_EMULATOR)); + } + + @Test + public void testStartWithEmulatorEnabled() { + props.put(ConnectorUtils.CPS_USE_EMULATOR, "true"); + props.put(ConnectorUtils.CPS_ENDPOINT, "localhost:8085"); + CloudPubSubSourceTask task = new CloudPubSubSourceTask(subscriber); + task.start(props); + assertEquals(CloudPubSubSourceTask.class, task.getClass()); + } + + @Test + public void testStartWithEmulatorDisabled() { + props.put(ConnectorUtils.CPS_USE_EMULATOR, "false"); + props.put(ConnectorUtils.CPS_ENDPOINT, "pubsub.googleapis.com:443"); + CloudPubSubSourceTask task = new CloudPubSubSourceTask(subscriber); + task.start(props); + assertEquals(CloudPubSubSourceTask.class, task.getClass()); + } + /** Tests when no messages are received from the Cloud Pub/Sub PullResponse. */ @Test public void testPollCaseWithNoMessages() throws Exception { From f73091d62d147d0c4bb0dd024d4062c439025052 Mon Sep 17 00:00:00 2001 From: Adalbert Makarovych Date: Wed, 27 May 2026 10:24:50 +0300 Subject: [PATCH 2/3] Removed unused import --- .../google/pubsub/kafka/source/CloudPubSubGRPCSubscriber.java | 1 - 1 file changed, 1 deletion(-) diff --git a/src/main/java/com/google/pubsub/kafka/source/CloudPubSubGRPCSubscriber.java b/src/main/java/com/google/pubsub/kafka/source/CloudPubSubGRPCSubscriber.java index b60e9b34..4cd2a40d 100644 --- a/src/main/java/com/google/pubsub/kafka/source/CloudPubSubGRPCSubscriber.java +++ b/src/main/java/com/google/pubsub/kafka/source/CloudPubSubGRPCSubscriber.java @@ -24,7 +24,6 @@ import com.google.cloud.pubsub.v1.stub.SubscriberStubSettings; import com.google.common.util.concurrent.MoreExecutors; import com.google.protobuf.Empty; -import com.google.pubsub.kafka.common.ConnectorUtils; import com.google.pubsub.v1.AcknowledgeRequest; import com.google.pubsub.v1.ProjectSubscriptionName; import com.google.pubsub.v1.PullRequest; From 5f868df342df802a27ea24e158f653b46b3bdc74 Mon Sep 17 00:00:00 2001 From: Adalbert Makarovych Date: Fri, 26 Jun 2026 12:34:02 +0300 Subject: [PATCH 3/3] Fixed lint failure --- .../google/pubsub/kafka/common/ConnectorUtils.java | 1 - .../source/CloudPubSubSourceConnectorTest.java | 14 ++++++++++++-- .../kafka/source/CloudPubSubSourceTaskTest.java | 1 - 3 files changed, 12 insertions(+), 4 deletions(-) diff --git a/src/main/java/com/google/pubsub/kafka/common/ConnectorUtils.java b/src/main/java/com/google/pubsub/kafka/common/ConnectorUtils.java index ece4e9a2..1d1b7d4d 100644 --- a/src/main/java/com/google/pubsub/kafka/common/ConnectorUtils.java +++ b/src/main/java/com/google/pubsub/kafka/common/ConnectorUtils.java @@ -59,7 +59,6 @@ public static synchronized ScheduledExecutorService getSystemExecutor() { return SYSTEM_EXECUTOR.get(); } - // Resolve the endpoint. When using the emulator, prefer PUBSUB_EMULATOR_HOST and fall back to // the configured cps.endpoint. public static String getPubsubEndpoint(boolean useEmulator, String cpsEndpoint) { diff --git a/src/test/java/com/google/pubsub/kafka/source/CloudPubSubSourceConnectorTest.java b/src/test/java/com/google/pubsub/kafka/source/CloudPubSubSourceConnectorTest.java index 2f3a814f..3e893899 100644 --- a/src/test/java/com/google/pubsub/kafka/source/CloudPubSubSourceConnectorTest.java +++ b/src/test/java/com/google/pubsub/kafka/source/CloudPubSubSourceConnectorTest.java @@ -55,7 +55,12 @@ public void setup() { public void testStartWhenSubscriptionNonexistant() { doThrow(new ConnectException("")) .when(connector) - .verifySubscription(anyString(), anyString(), any(ConnectorCredentialsProvider.class), anyString(), anyBoolean()); + .verifySubscription( + anyString(), + anyString(), + any(ConnectorCredentialsProvider.class), + anyString(), + anyBoolean()); connector.start(props); } @@ -68,7 +73,12 @@ public void testStartWhenRequiredConfigMissing() { public void testTaskConfigs() { doNothing() .when(connector) - .verifySubscription(anyString(), anyString(), any(ConnectorCredentialsProvider.class), anyString(), anyBoolean()); + .verifySubscription( + anyString(), + anyString(), + any(ConnectorCredentialsProvider.class), + anyString(), + anyBoolean()); connector.start(props); List> taskConfigs = connector.taskConfigs(NUM_TASKS); assertEquals(taskConfigs.size(), NUM_TASKS); diff --git a/src/test/java/com/google/pubsub/kafka/source/CloudPubSubSourceTaskTest.java b/src/test/java/com/google/pubsub/kafka/source/CloudPubSubSourceTaskTest.java index e0a86763..3d83ca62 100644 --- a/src/test/java/com/google/pubsub/kafka/source/CloudPubSubSourceTaskTest.java +++ b/src/test/java/com/google/pubsub/kafka/source/CloudPubSubSourceTaskTest.java @@ -37,7 +37,6 @@ import java.util.HashMap; import java.util.List; import java.util.Map; - import org.apache.kafka.common.config.ConfigDef; import org.apache.kafka.connect.data.Field; import org.apache.kafka.connect.data.Schema;