Package io.aiven.commons.kafka.connector.source

Class AbstractSourceTask

java.lang.Object
org.apache.kafka.connect.source.SourceTask
io.aiven.commons.kafka.connector.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
      A null list representing no values in the polling functionality.

    • 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 latency.

  • Constructor Details

    • AbstractSourceTask

      protected AbstractSourceTask()
      Constructor.

  • Method Details

    • getIterator

      protected abstract EvolvingSourceRecordIterator getIterator(SourceCommonConfig 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 SourceCommonConfig instance.
      Returns:
      The iterator of SourceRecords.

    • configure

      protected abstract SourceCommonConfig configure(Map<String,String> props, OffsetManager offsetManager)
      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.
      offsetManager - the OffsetManager to use.
      Returns:
      A SourceCommonConfig based 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

    • lastEvolution

      protected EvolvingSourceRecord lastEvolution(EvolvingSourceRecord evolvingSourceRecord)
      An insertion point for the last evolution step in the EvolvingSourceRecord. This method is called just before the EvolvingSourceRecord is converted into a SourceRecord. This is the last point to modify
      • Key schema and/or value
      • Value schema and/or value
      • Offset manager entry
      The default implementation makes no changes.
      Parameters:
      evolvingSourceRecord - the record to modify.
      Returns:
      the modified record.

    • 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 when stopping.