diff --git a/checkstyle/import-control.xml b/checkstyle/import-control.xml index 238366dcac5..e3f4f84c6be 100644 --- a/checkstyle/import-control.xml +++ b/checkstyle/import-control.xml @@ -109,6 +109,7 @@ + @@ -124,7 +125,6 @@ - diff --git a/copycat/api/src/main/java/org/apache/kafka/copycat/connector/Connector.java b/copycat/api/src/main/java/org/apache/kafka/copycat/connector/Connector.java index b39a8be4021..7127ff71222 100644 --- a/copycat/api/src/main/java/org/apache/kafka/copycat/connector/Connector.java +++ b/copycat/api/src/main/java/org/apache/kafka/copycat/connector/Connector.java @@ -17,6 +17,7 @@ package org.apache.kafka.copycat.connector; +import org.apache.kafka.common.annotation.InterfaceStability; import org.apache.kafka.copycat.errors.CopycatException; import java.util.List; @@ -39,6 +40,7 @@ import java.util.Properties; * Tasks. *

*/ +@InterfaceStability.Unstable public abstract class Connector { protected ConnectorContext context; diff --git a/copycat/api/src/main/java/org/apache/kafka/copycat/connector/ConnectorContext.java b/copycat/api/src/main/java/org/apache/kafka/copycat/connector/ConnectorContext.java index 06ac81d93d8..ecba69af8a4 100644 --- a/copycat/api/src/main/java/org/apache/kafka/copycat/connector/ConnectorContext.java +++ b/copycat/api/src/main/java/org/apache/kafka/copycat/connector/ConnectorContext.java @@ -17,9 +17,12 @@ package org.apache.kafka.copycat.connector; +import org.apache.kafka.common.annotation.InterfaceStability; + /** * ConnectorContext allows Connectors to proactively interact with the Copycat runtime. */ +@InterfaceStability.Unstable public interface ConnectorContext { /** * Requests that the runtime reconfigure the Tasks for this source. This should be used to diff --git a/copycat/api/src/main/java/org/apache/kafka/copycat/connector/CopycatRecord.java b/copycat/api/src/main/java/org/apache/kafka/copycat/connector/CopycatRecord.java index 3060c6e4e7c..576904ad67f 100644 --- a/copycat/api/src/main/java/org/apache/kafka/copycat/connector/CopycatRecord.java +++ b/copycat/api/src/main/java/org/apache/kafka/copycat/connector/CopycatRecord.java @@ -17,6 +17,8 @@ package org.apache.kafka.copycat.connector; +import org.apache.kafka.common.annotation.InterfaceStability; + /** *

* Base class for records containing data to be copied to/from Kafka. This corresponds closely to @@ -25,6 +27,7 @@ package org.apache.kafka.copycat.connector; * notion of offset, it is not included here because they differ in type. *

*/ +@InterfaceStability.Unstable public abstract class CopycatRecord { private final String topic; private final Integer kafkaPartition; diff --git a/copycat/api/src/main/java/org/apache/kafka/copycat/connector/Task.java b/copycat/api/src/main/java/org/apache/kafka/copycat/connector/Task.java index 1c5efdd59a4..40fbe18dab5 100644 --- a/copycat/api/src/main/java/org/apache/kafka/copycat/connector/Task.java +++ b/copycat/api/src/main/java/org/apache/kafka/copycat/connector/Task.java @@ -17,6 +17,7 @@ package org.apache.kafka.copycat.connector; +import org.apache.kafka.common.annotation.InterfaceStability; import org.apache.kafka.copycat.errors.CopycatException; import java.util.Properties; @@ -34,6 +35,7 @@ import java.util.Properties; * {@link org.apache.kafka.copycat.sink.SinkTask}. *

*/ +@InterfaceStability.Unstable public interface Task { /** * Start the Task diff --git a/copycat/api/src/main/java/org/apache/kafka/copycat/errors/CopycatException.java b/copycat/api/src/main/java/org/apache/kafka/copycat/errors/CopycatException.java index ac46250f8c8..5d37557dd32 100644 --- a/copycat/api/src/main/java/org/apache/kafka/copycat/errors/CopycatException.java +++ b/copycat/api/src/main/java/org/apache/kafka/copycat/errors/CopycatException.java @@ -17,9 +17,12 @@ package org.apache.kafka.copycat.errors; +import org.apache.kafka.common.annotation.InterfaceStability; + /** * CopycatException is the top-level exception type generated by Copycat and connectors. */ +@InterfaceStability.Unstable public class CopycatException extends Exception { public CopycatException(String s) { diff --git a/copycat/api/src/main/java/org/apache/kafka/copycat/errors/CopycatRuntimeException.java b/copycat/api/src/main/java/org/apache/kafka/copycat/errors/CopycatRuntimeException.java index 126526cda6b..61e94c5eac8 100644 --- a/copycat/api/src/main/java/org/apache/kafka/copycat/errors/CopycatRuntimeException.java +++ b/copycat/api/src/main/java/org/apache/kafka/copycat/errors/CopycatRuntimeException.java @@ -17,6 +17,9 @@ package org.apache.kafka.copycat.errors; +import org.apache.kafka.common.annotation.InterfaceStability; + +@InterfaceStability.Unstable public class CopycatRuntimeException extends RuntimeException { public CopycatRuntimeException(String s) { super(s); diff --git a/copycat/api/src/main/java/org/apache/kafka/copycat/sink/SinkConnector.java b/copycat/api/src/main/java/org/apache/kafka/copycat/sink/SinkConnector.java index 2755d17033e..fb2e694df99 100644 --- a/copycat/api/src/main/java/org/apache/kafka/copycat/sink/SinkConnector.java +++ b/copycat/api/src/main/java/org/apache/kafka/copycat/sink/SinkConnector.java @@ -17,11 +17,13 @@ package org.apache.kafka.copycat.sink; +import org.apache.kafka.common.annotation.InterfaceStability; import org.apache.kafka.copycat.connector.Connector; /** * SinkConnectors implement the Connector interface to send Kafka data to another system. */ +@InterfaceStability.Unstable public abstract class SinkConnector extends Connector { /** diff --git a/copycat/api/src/main/java/org/apache/kafka/copycat/sink/SinkRecord.java b/copycat/api/src/main/java/org/apache/kafka/copycat/sink/SinkRecord.java index f0d1387030b..e3775b358af 100644 --- a/copycat/api/src/main/java/org/apache/kafka/copycat/sink/SinkRecord.java +++ b/copycat/api/src/main/java/org/apache/kafka/copycat/sink/SinkRecord.java @@ -17,6 +17,7 @@ package org.apache.kafka.copycat.sink; +import org.apache.kafka.common.annotation.InterfaceStability; import org.apache.kafka.copycat.connector.CopycatRecord; /** @@ -24,6 +25,7 @@ import org.apache.kafka.copycat.connector.CopycatRecord; * the record in the Kafka topic-partition in addition to the standard fields. This information * should be used by the SinkTask to coordinate kafkaOffset commits. */ +@InterfaceStability.Unstable public class SinkRecord extends CopycatRecord { private final long kafkaOffset; diff --git a/copycat/api/src/main/java/org/apache/kafka/copycat/sink/SinkTask.java b/copycat/api/src/main/java/org/apache/kafka/copycat/sink/SinkTask.java index 46c4968a55b..6eb6fd54a5c 100644 --- a/copycat/api/src/main/java/org/apache/kafka/copycat/sink/SinkTask.java +++ b/copycat/api/src/main/java/org/apache/kafka/copycat/sink/SinkTask.java @@ -17,6 +17,7 @@ package org.apache.kafka.copycat.sink; import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.annotation.InterfaceStability; import org.apache.kafka.copycat.connector.Task; import org.apache.kafka.copycat.errors.CopycatException; @@ -28,6 +29,7 @@ import java.util.Map; * addition to the basic {@link #put} interface, SinkTasks must also implement {@link #flush} * to support offset commits. */ +@InterfaceStability.Unstable public abstract class SinkTask implements Task { /** diff --git a/copycat/api/src/main/java/org/apache/kafka/copycat/sink/SinkTaskContext.java b/copycat/api/src/main/java/org/apache/kafka/copycat/sink/SinkTaskContext.java index f4231098805..7cc6109622b 100644 --- a/copycat/api/src/main/java/org/apache/kafka/copycat/sink/SinkTaskContext.java +++ b/copycat/api/src/main/java/org/apache/kafka/copycat/sink/SinkTaskContext.java @@ -18,6 +18,7 @@ package org.apache.kafka.copycat.sink; import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.annotation.InterfaceStability; import java.util.HashMap; import java.util.Map; @@ -25,6 +26,7 @@ import java.util.Map; /** * Context passed to SinkTasks, allowing them to access utilities in the copycat runtime. */ +@InterfaceStability.Unstable public abstract class SinkTaskContext { private Map offsets; diff --git a/copycat/api/src/main/java/org/apache/kafka/copycat/source/SourceConnector.java b/copycat/api/src/main/java/org/apache/kafka/copycat/source/SourceConnector.java index 10fd07f4184..7258cdfb826 100644 --- a/copycat/api/src/main/java/org/apache/kafka/copycat/source/SourceConnector.java +++ b/copycat/api/src/main/java/org/apache/kafka/copycat/source/SourceConnector.java @@ -16,12 +16,14 @@ **/ package org.apache.kafka.copycat.source; +import org.apache.kafka.common.annotation.InterfaceStability; import org.apache.kafka.copycat.connector.Connector; /** * SourceConnectors implement the connector interface to pull data from another system and send * it to Kafka. */ +@InterfaceStability.Unstable public abstract class SourceConnector extends Connector { } diff --git a/copycat/api/src/main/java/org/apache/kafka/copycat/source/SourceRecord.java b/copycat/api/src/main/java/org/apache/kafka/copycat/source/SourceRecord.java index cfe0dcf85fc..2085f6604f6 100644 --- a/copycat/api/src/main/java/org/apache/kafka/copycat/source/SourceRecord.java +++ b/copycat/api/src/main/java/org/apache/kafka/copycat/source/SourceRecord.java @@ -17,6 +17,7 @@ package org.apache.kafka.copycat.source; +import org.apache.kafka.common.annotation.InterfaceStability; import org.apache.kafka.copycat.connector.CopycatRecord; /** @@ -37,6 +38,7 @@ import org.apache.kafka.copycat.connector.CopycatRecord; * "table_name"} and the sourceOffset as a Long containing the timestamp of the row. *

*/ +@InterfaceStability.Unstable public class SourceRecord extends CopycatRecord { private final Object sourcePartition; private final Object sourceOffset; diff --git a/copycat/api/src/main/java/org/apache/kafka/copycat/source/SourceTask.java b/copycat/api/src/main/java/org/apache/kafka/copycat/source/SourceTask.java index a5a0671a6e9..1e1da345979 100644 --- a/copycat/api/src/main/java/org/apache/kafka/copycat/source/SourceTask.java +++ b/copycat/api/src/main/java/org/apache/kafka/copycat/source/SourceTask.java @@ -17,6 +17,7 @@ package org.apache.kafka.copycat.source; +import org.apache.kafka.common.annotation.InterfaceStability; import org.apache.kafka.copycat.connector.Task; import java.util.List; @@ -24,6 +25,7 @@ import java.util.List; /** * SourceTask is a Task that pulls records from another system for storage in Kafka. */ +@InterfaceStability.Unstable public abstract class SourceTask implements Task { protected SourceTaskContext context; diff --git a/copycat/api/src/main/java/org/apache/kafka/copycat/source/SourceTaskContext.java b/copycat/api/src/main/java/org/apache/kafka/copycat/source/SourceTaskContext.java index 29e0b3ea9b9..d52fd624c79 100644 --- a/copycat/api/src/main/java/org/apache/kafka/copycat/source/SourceTaskContext.java +++ b/copycat/api/src/main/java/org/apache/kafka/copycat/source/SourceTaskContext.java @@ -16,12 +16,14 @@ **/ package org.apache.kafka.copycat.source; +import org.apache.kafka.common.annotation.InterfaceStability; import org.apache.kafka.copycat.storage.OffsetStorageReader; /** * SourceTaskContext is provided to SourceTasks to allow them to interact with the underlying * runtime. */ +@InterfaceStability.Unstable public class SourceTaskContext { private final OffsetStorageReader reader; diff --git a/copycat/api/src/main/java/org/apache/kafka/copycat/storage/Converter.java b/copycat/api/src/main/java/org/apache/kafka/copycat/storage/Converter.java index 9f521ed77c1..c50aee7c63f 100644 --- a/copycat/api/src/main/java/org/apache/kafka/copycat/storage/Converter.java +++ b/copycat/api/src/main/java/org/apache/kafka/copycat/storage/Converter.java @@ -17,6 +17,8 @@ package org.apache.kafka.copycat.storage; +import org.apache.kafka.common.annotation.InterfaceStability; + /** * The Converter interface provides support for translating between Copycat's runtime data format * and the "native" runtime format used by the serialization layer. This is used to translate @@ -24,6 +26,7 @@ package org.apache.kafka.copycat.storage; * component -- the producer or consumer serializer or deserializer for records or a Copycat * serializer or deserializer for offsets. */ +@InterfaceStability.Unstable public interface Converter { /** diff --git a/copycat/api/src/main/java/org/apache/kafka/copycat/storage/OffsetStorageReader.java b/copycat/api/src/main/java/org/apache/kafka/copycat/storage/OffsetStorageReader.java index 3501cb533e3..785660d28d3 100644 --- a/copycat/api/src/main/java/org/apache/kafka/copycat/storage/OffsetStorageReader.java +++ b/copycat/api/src/main/java/org/apache/kafka/copycat/storage/OffsetStorageReader.java @@ -17,6 +17,8 @@ package org.apache.kafka.copycat.storage; +import org.apache.kafka.common.annotation.InterfaceStability; + import java.util.Collection; import java.util.Map; @@ -25,6 +27,7 @@ import java.util.Map; * connectors to determine offsets to start consuming data from. This is most commonly used during * initialization of a task, but can also be used during runtime, e.g. when reconfiguring a task. */ +@InterfaceStability.Unstable public interface OffsetStorageReader { /** * Get the offset for the specified partition. If the data isn't already available locally, this diff --git a/copycat/api/src/main/java/org/apache/kafka/copycat/util/ConnectorUtils.java b/copycat/api/src/main/java/org/apache/kafka/copycat/util/ConnectorUtils.java index 2a8b5325058..f9dd53ad4a1 100644 --- a/copycat/api/src/main/java/org/apache/kafka/copycat/util/ConnectorUtils.java +++ b/copycat/api/src/main/java/org/apache/kafka/copycat/util/ConnectorUtils.java @@ -17,6 +17,8 @@ package org.apache.kafka.copycat.util; +import org.apache.kafka.common.annotation.InterfaceStability; + import java.util.ArrayList; import java.util.List; @@ -24,6 +26,7 @@ import java.util.List; * Utilities that connector implementations might find useful. Contains common building blocks * for writing connectors. */ +@InterfaceStability.Unstable public class ConnectorUtils { /** * Given a list of elements and a target number of groups, generates list of groups of diff --git a/copycat/runtime/src/main/java/org/apache/kafka/copycat/cli/Copycat.java b/copycat/runtime/src/main/java/org/apache/kafka/copycat/cli/Copycat.java index 42be14da823..32b785e5cab 100644 --- a/copycat/runtime/src/main/java/org/apache/kafka/copycat/cli/Copycat.java +++ b/copycat/runtime/src/main/java/org/apache/kafka/copycat/cli/Copycat.java @@ -17,6 +17,7 @@ package org.apache.kafka.copycat.cli; +import org.apache.kafka.common.annotation.InterfaceStability; import org.apache.kafka.common.config.ConfigException; import org.apache.kafka.common.utils.Utils; import org.apache.kafka.copycat.runtime.Coordinator; @@ -40,6 +41,7 @@ import java.util.Properties; * fault tolerant by overriding the settings to use file storage for both. *

*/ +@InterfaceStability.Unstable public class Copycat { private static final Logger log = LoggerFactory.getLogger(Copycat.class); diff --git a/copycat/runtime/src/main/java/org/apache/kafka/copycat/cli/CopycatConfig.java b/copycat/runtime/src/main/java/org/apache/kafka/copycat/cli/CopycatConfig.java index 74cb24374d3..0584ace4be3 100644 --- a/copycat/runtime/src/main/java/org/apache/kafka/copycat/cli/CopycatConfig.java +++ b/copycat/runtime/src/main/java/org/apache/kafka/copycat/cli/CopycatConfig.java @@ -17,6 +17,7 @@ package org.apache.kafka.copycat.cli; +import org.apache.kafka.common.annotation.InterfaceStability; import org.apache.kafka.common.config.AbstractConfig; import org.apache.kafka.common.config.ConfigDef; import org.apache.kafka.common.config.ConfigDef.Importance; @@ -24,6 +25,7 @@ import org.apache.kafka.common.config.ConfigDef.Type; import java.util.Properties; +@InterfaceStability.Unstable public class CopycatConfig extends AbstractConfig { public static final String WORKER_PROPERTIES_CONFIG = "worker-config"; @@ -53,11 +55,8 @@ public class CopycatConfig extends AbstractConfig { Importance.HIGH, DELETE_CONNECTORS_CONFIG_DOC); } - private Properties originalProperties; - public CopycatConfig(Properties props) { super(config, props); - this.originalProperties = props; } /** diff --git a/copycat/runtime/src/main/java/org/apache/kafka/copycat/cli/WorkerConfig.java b/copycat/runtime/src/main/java/org/apache/kafka/copycat/cli/WorkerConfig.java index 62004928e3e..577d708dab9 100644 --- a/copycat/runtime/src/main/java/org/apache/kafka/copycat/cli/WorkerConfig.java +++ b/copycat/runtime/src/main/java/org/apache/kafka/copycat/cli/WorkerConfig.java @@ -17,6 +17,7 @@ package org.apache.kafka.copycat.cli; +import org.apache.kafka.common.annotation.InterfaceStability; import org.apache.kafka.common.config.AbstractConfig; import org.apache.kafka.common.config.ConfigDef; import org.apache.kafka.common.config.ConfigDef.Importance; @@ -28,6 +29,7 @@ import java.util.Set; /** * Configuration for standalone workers. */ +@InterfaceStability.Unstable public class WorkerConfig extends AbstractConfig { public static final String CLUSTER_CONFIG = "cluster"; diff --git a/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/Copycat.java b/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/Copycat.java index 15030cc339f..87047e8af2c 100644 --- a/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/Copycat.java +++ b/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/Copycat.java @@ -17,6 +17,7 @@ package org.apache.kafka.copycat.runtime; +import org.apache.kafka.common.annotation.InterfaceStability; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -27,6 +28,7 @@ import java.util.concurrent.atomic.AtomicBoolean; * This class ties together all the components of a Copycat process (coordinator, worker, * storage, command interface), managing their lifecycle. */ +@InterfaceStability.Unstable public class Copycat { private static final Logger log = LoggerFactory.getLogger(Copycat.class); diff --git a/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/SinkTaskContextImpl.java b/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/SinkTaskContextImpl.java index e43c0af66ac..f47c984b6af 100644 --- a/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/SinkTaskContextImpl.java +++ b/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/SinkTaskContextImpl.java @@ -19,6 +19,6 @@ package org.apache.kafka.copycat.runtime; import org.apache.kafka.copycat.sink.SinkTaskContext; -public class SinkTaskContextImpl extends SinkTaskContext { +class SinkTaskContextImpl extends SinkTaskContext { } diff --git a/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/SourceTaskOffsetCommitter.java b/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/SourceTaskOffsetCommitter.java index 8bda25e6432..953cfa5f148 100644 --- a/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/SourceTaskOffsetCommitter.java +++ b/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/SourceTaskOffsetCommitter.java @@ -44,7 +44,7 @@ import java.util.concurrent.TimeUnit; * The current implementation uses a single thread to process commits and *

*/ -public class SourceTaskOffsetCommitter { +class SourceTaskOffsetCommitter { private static final Logger log = LoggerFactory.getLogger(SourceTaskOffsetCommitter.class); private Time time; diff --git a/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/WorkerSinkTask.java b/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/WorkerSinkTask.java index 05a212ae4d7..b1893b9ba7a 100644 --- a/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/WorkerSinkTask.java +++ b/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/WorkerSinkTask.java @@ -38,7 +38,7 @@ import java.util.concurrent.TimeUnit; /** * WorkerTask that uses a SinkTask to export data from Kafka. */ -public class WorkerSinkTask implements WorkerTask { +class WorkerSinkTask implements WorkerTask { private static final Logger log = LoggerFactory.getLogger(WorkerSinkTask.class); private final ConnectorTaskId id; diff --git a/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/WorkerSourceTask.java b/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/WorkerSourceTask.java index b7c70371cf6..a23fe090802 100644 --- a/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/WorkerSourceTask.java +++ b/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/WorkerSourceTask.java @@ -46,7 +46,7 @@ import java.util.concurrent.TimeoutException; /** * WorkerTask that uses a SourceTask to ingest data into Kafka. */ -public class WorkerSourceTask implements WorkerTask { +class WorkerSourceTask implements WorkerTask { private static final Logger log = LoggerFactory.getLogger(WorkerSourceTask.class); private ConnectorTaskId id; diff --git a/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/WorkerTask.java b/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/WorkerTask.java index e54f00942a4..1cdf597699f 100644 --- a/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/WorkerTask.java +++ b/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/WorkerTask.java @@ -26,7 +26,7 @@ import java.util.Properties; * used by {@link Worker} to manage the tasks. Implementations combine a user-specified Task with * Kafka to create a data flow. */ -public interface WorkerTask { +interface WorkerTask { /** * Start the Task * @param props initial configuration diff --git a/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/standalone/ConfigStorage.java b/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/standalone/ConfigStorage.java index 05c1f9691a2..f7bf31b3d93 100644 --- a/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/standalone/ConfigStorage.java +++ b/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/standalone/ConfigStorage.java @@ -24,7 +24,7 @@ import java.util.Properties; * Interface used by StandaloneController to store configuration data for jobs. To be fault * tolerant, all data required to resume jobs is stored here. */ -public interface ConfigStorage { +interface ConfigStorage { /** * Configure this storage engine. diff --git a/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/standalone/StandaloneConnectorContext.java b/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/standalone/StandaloneConnectorContext.java index 5d4473208bb..eb319a01b29 100644 --- a/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/standalone/StandaloneConnectorContext.java +++ b/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/standalone/StandaloneConnectorContext.java @@ -23,7 +23,7 @@ import org.apache.kafka.copycat.connector.ConnectorContext; * ConnectorContext for use with the StandaloneCoordinator, which maintains all connectors and tasks * in a single process. */ -public class StandaloneConnectorContext implements ConnectorContext { +class StandaloneConnectorContext implements ConnectorContext { private StandaloneCoordinator coordinator; private String connectorName;