Package io.debezium.connector.common

Class BaseSourceTask<P extends Partition,O extends OffsetContext>

java.lang.Object
org.apache.kafka.connect.source.SourceTask
io.debezium.connector.common.BaseSourceTask<P,O>
All Implemented Interfaces:
org.apache.kafka.connect.connector.Task
public abstract class BaseSourceTask<P extends Partition,O extends OffsetContext> extends org.apache.kafka.connect.source.SourceTask
Base class for Debezium's CDC SourceTask implementations. Provides functionality common to all connectors, such as validation of the configuration.
Author:
Gunnar Morling

  • Field Details

    • LOGGER

      private static final org.slf4j.Logger LOGGER

    • INITIAL_POLL_PERIOD_IN_MILLIS

      private static final Duration INITIAL_POLL_PERIOD_IN_MILLIS

    • MAX_POLL_PERIOD_IN_MILLIS

      private static final Duration MAX_POLL_PERIOD_IN_MILLIS

    • config

      private io.debezium.config.Configuration config

    • signalChannels

      private List<SignalChannelReader> signalChannels

    • cdcSourceTaskContext

      private CdcSourceTaskContext<? extends CommonConnectorConfig> cdcSourceTaskContext

    • state

      private final AtomicReference<io.debezium.connector.common.DebeziumTaskState> state

    • stateLock

      private final ReentrantLock stateLock
      Used to ensure that start(), stop() and commitRecord() calls are serialized.

    • restartDelay

      private volatile io.debezium.util.ElapsedTimeStrategy restartDelay

    • coordinator

      protected ChangeEventSourceCoordinator<P extends Partition,O extends OffsetContext> coordinator
      The change event source coordinator for those connectors adhering to the new framework structure, null for legacy-style connectors.

    • lastOffsets

      private final Map<Map<String,?>,Map<String,?>> lastOffsets
      The latest offsets that have been acknowledged by the Kafka producer. Will be acknowledged with the source database in commit() (which may be a no-op depending on the connector).

    • retriableRestartWait

      private Duration retriableRestartWait

    • pollOutputDelay

      private final io.debezium.util.ElapsedTimeStrategy pollOutputDelay

    • clock

      private final io.debezium.util.Clock clock

    • previousOutputInstant

      private Instant previousOutputInstant

    • previousOutputBatchSize

      private int previousOutputBatchSize

    • shouldPerformCommit

      protected final AtomicBoolean shouldPerformCommit

    • availableSignalChannels

      private final ServiceLoader<SignalChannelReader> availableSignalChannels

    • notificationChannels

      private final List<NotificationChannel> notificationChannels

    • offsetLoadedInPast

      private boolean offsetLoadedInPast
      A flag to record whether the offsets stored in the offset store are loaded for the first time. This is typically used to reduce logging in case a connector like PostgreSQL reads offsets not only on connector startup but repeatedly during execution time too.

  • Constructor Details

    • BaseSourceTask

      protected BaseSourceTask()

  • Method Details

    • validateSchemaHistory

      protected void validateSchemaHistory(CommonConnectorConfig config, LogPositionValidator logPositionValidator, Offsets<P,O> previousOffsets, DatabaseSchema schema, io.debezium.spi.snapshot.Snapshotter snapshotter)

    • isLogPositionAvailable

      public boolean isLogPositionAvailable(LogPositionValidator logPositionValidator, Partition partition, OffsetContext offsetContext, CommonConnectorConfig config)

    • preStart

      public abstract CdcSourceTaskContext<? extends CommonConnectorConfig> preStart(io.debezium.config.Configuration config)

    • start

      public final void start(Map<String,String> props)
      Specified by:
      start in interface org.apache.kafka.connect.connector.Task
      Specified by:
      start in class org.apache.kafka.connect.source.SourceTask

    • getAvailableSignalChannels

      public List<SignalChannelReader> getAvailableSignalChannels()
      Returns the available signal channels.

      The signal channels are loaded using the ServiceLoader mechanism and cached for the lifetime of the task.

      Returns:
      list of loaded signal channels

    • getAvailableSignalChannelWriter

      public Optional<? extends SignalChannelWriter> getAvailableSignalChannelWriter()
      Returns the first available signal channel writer
      Returns:
      the first available signal channel writer empty optional if not available

    • withMaskedSensitiveOptions

      protected io.debezium.config.Configuration withMaskedSensitiveOptions(io.debezium.config.Configuration config)

    • getMaskedConfigurationMap

      protected Map<String,String> getMaskedConfigurationMap(Map<String,String> props)

    • start

      protected abstract ChangeEventSourceCoordinator<P,O> start(io.debezium.config.Configuration config)
      Called when starting this source task. This method can throw a RetriableException to indicate that the task should attempt to retry the start later.
      Parameters:
      config - the task configuration; implementations should wrap it in a dedicated implementation of CommonConnectorConfig and work with typed access to configuration properties that way

    • connectorName

      protected abstract String connectorName()

    • poll

      public final List<org.apache.kafka.connect.source.SourceRecord> poll() throws InterruptedException
      Specified by:
      poll in class org.apache.kafka.connect.source.SourceTask
      Throws:
      InterruptedException

    • logStatistics

      protected void logStatistics(List<org.apache.kafka.connect.source.SourceRecord> records)

    • updateLastOffset

      private void updateLastOffset(Map<String,?> partition, Map<String,?> lastOffset)

    • resetErrorHandlerRetriesIfNeeded

      protected void resetErrorHandlerRetriesIfNeeded(List<org.apache.kafka.connect.source.SourceRecord> records)
      Should be called to reset the error handler's retry counter upon a successful poll or when known that the connector task has recovered from a previous failure state.

    • containsChangeDataMessages

      protected boolean containsChangeDataMessages(List<org.apache.kafka.connect.source.SourceRecord> records)

    • doPoll

      protected abstract List<org.apache.kafka.connect.source.SourceRecord> doPoll() throws InterruptedException
      Returns the next batch of source records, if any are available.
      Throws:
      InterruptedException

    • getErrorHandler

      protected abstract Optional<ErrorHandler> getErrorHandler()

    • startIfNeededAndPossible

      private boolean startIfNeededAndPossible() throws InterruptedException
      Starts this connector in case it has been stopped after a retriable error, and the backoff period has passed.
      Throws:
      InterruptedException

    • stop

      public final void stop()
      Specified by:
      stop in interface org.apache.kafka.connect.connector.Task
      Specified by:
      stop in class org.apache.kafka.connect.source.SourceTask

    • stop

      private void stop(boolean restart)

    • doStop

      protected abstract void doStop()

    • commitRecord

      public void commitRecord(org.apache.kafka.connect.source.SourceRecord record, org.apache.kafka.clients.producer.RecordMetadata metadata) throws InterruptedException
      Overrides:
      commitRecord in class org.apache.kafka.connect.source.SourceTask
      Throws:
      InterruptedException

    • commit

      public void commit() throws InterruptedException
      Overrides:
      commit in class org.apache.kafka.connect.source.SourceTask
      Throws:
      InterruptedException

    • performCommit

      public void performCommit()

    • getAllConfigurationFields

      protected abstract Iterable<io.debezium.config.Field> getAllConfigurationFields()
      Returns all configuration Field supported by this source task.

    • getPreviousOffsets

      protected Offsets<P,O> getPreviousOffsets(Partition.Provider<P> provider, OffsetContext.Loader<O> loader)
      Loads the connector's persistent offsets (if present) via the given loader.

    • setTaskState

      private void setTaskState(io.debezium.connector.common.DebeziumTaskState newState)
      Sets the new state for the task. The caller must be holding stateLock lock.
      Parameters:
      newState -

    • getTaskState

      @VisibleForTesting public io.debezium.connector.common.DebeziumTaskState getTaskState()

    • getNotificationChannels

      public List<NotificationChannel> getNotificationChannels()

    • registerServiceProviders

      protected void registerServiceProviders(ServiceRegistry serviceRegistry)