Package io.aiven.kafka.connect.common.source

Class AbstractSourceTask

java.lang.Object
org.apache.kafka.connect.source.SourceTask
io.aiven.kafka.connect.common.source.AbstractSourceTask
All Implemented Interfaces:
org.apache.kafka.connect.connector.Task
public abstract class AbstractSourceTask extends org.apache.kafka.connect.source.SourceTask
This class handles extracting records from an iterator and returning them to Kafka. It uses an exponential backoff with jitter to reduce the number of calls to the backend when there is no data. This solution:
  • When polled this implementation moves available records from the SourceRecord iterator to the return array.
  • if there are no records
    • poll() will return null.
    • The poll will delay no more than approx 5 seconds.
  • Upto maxPollRecords will be sent in a single poll request
  • When the connector is stopped any collected records are returned to kafka before stopping.

  • Field Details

    • NULL_RESULT

      public static final List<org.apache.kafka.connect.source.SourceRecord> NULL_RESULT

    • MAX_POLL_TIME

      public static final Duration MAX_POLL_TIME
      The maximum time to spend polling. This is set to 4 seconds as 5 seconds is the kafka limit that is allotted to a system for shutdown, and this allows the polling and iterator to smoothly shutdown accounting for latentcy.

  • Constructor Details

    • AbstractSourceTask

      protected AbstractSourceTask(org.slf4j.Logger logger)
      Constructor.
      Parameters:
      logger - the logger to use.

  • Method Details

    • getIterator

      protected abstract Iterator<org.apache.kafka.connect.source.SourceRecord> getIterator(AbstractSourceTask.BackoffConfig config)
      Gets the iterator of SourceRecords. The iterator that SourceRecords are extracted from for a poll event. When this iterator runs out of records it should attempt to reset and read more records from the backend on the next hasNext() call. In this way it should detect when new data has been added to the backend and continue processing.

      This method should handle any backend exception that can be retried. Any runtime exceptions that are thrown when this iterator executes may cause the task to abort.

      Parameters:
      config - the configuration for the Backoff.
      Returns:
      The iterator of SourceRecords.

    • configure

      protected abstract SourceCommonConfig configure(Map<String,String> props)
      Called by start(java.util.Map<java.lang.String, java.lang.String>) to allows the concrete implementation to configure itself based on properties.
      Parameters:
      props - the properties to use for configuration.

    • 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

    • stillPolling

      protected final boolean stillPolling()
      Returns true if the connector is not stopped and the timer has not expired.
      Returns:
      true if the connector is not stopped and the timer has not expired.

    • poll

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

    • 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

    • isRunning

      public final boolean isRunning()
      Returns the running state of the task.
      Returns:
      true if the connector is running, false otherwise.

    • closeResources

      protected abstract void closeResources()
      Close any resources the source has open. Called by the IteratorRunnable when it is stopping.