Timer: Throttling HttpRequest

Controls the rate at which HTTP requests are performed by informing the task, how long until the next execution is due.

http.timer

```java public interface Timer extends Configurable {

Long getRemainingMillis();

default void reset(Instant lastZero) {
    // Do nothing
}

} `` * Type:Class* Default:com.github.castorm.kafka.connect.timer.AdaptableIntervalTimer* Available implementations: *com.github.castorm.kafka.connect.timer.FixedIntervalTimer*com.github.castorm.kafka.connect.timer.AdaptableIntervalTimer`

Throttling HttpRequest with FixedIntervalThrottler

Throttles rate of requests based on a fixed interval.

http.timer.interval.millis

Interval in between requests * Type: Long * Default: 60000

Throttling HttpRequests with AdaptableIntervalThrottler

Throttles rate of requests based on a fixed interval. It has, however, two modes of operation, with two different intervals: * Up to date No new records in last poll, or there were new records, but "recently" created (shorter than interval) * Catching up There were new records in last poll, but they were created "long ago" (longer than interval)

http.timer.interval.millis

Interval in between requests when up-to-date

• Type: Long
• Default: 60000

http.timer.catchup.interval.millis

Interval in between requests when catching up * Type: Long * Default: 30000

HttpRequestFactory: Creating a HttpRequest

The first thing our connector will need to do is creating a HttpRequest.

http.request.factory

```java public interface HttpRequestFactory extends Configurable {

HttpRequest createRequest(Offset offset);

} `` * Type:Class* Default:com.github.castorm.kafka.connect.http.request.template.TemplateHttpRequestFactory* Available implementations: *com.github.castorm.kafka.connect.http.request.template.TemplateHttpRequestFactory`

http.offset.initial

Initial offset, comma separated list of pairs. * Example: property1=value1, property2=value2 * Type: String * Default: ""

Creating a HttpRequest with TemplateHttpRequestFactory

This HttpRequestFactory is based on template resolution.

http.request.method

Http method to use in the request. * Type: String * Default: GET

http.request.url

Http url to use in the request. * Required * Type: String

http.request.headers

Http headers to use in the request, , separated list of : separated pairs. * Example: Name: Value, Name2: Value2 * Type: String * Default: ""

http.request.params

Http query parameters to use in the request, & separated list of = separated pairs. * Example: name=value & name2=value2 * Type: String * Default: ""

http.request.body

Http body to use in the request. * Type: String * Default: ""

http.request.template.factory

```java public interface TemplateFactory {

Template create(String template);

}

public interface Template {

String apply(Offset offset);

} `` Class responsible for creating the templates that will be used on every request. * Type:Class* Default:com.github.castorm.kafka.connect.http.request.template.freemarker.BackwardsCompatibleFreeMarkerTemplateFactory* Available implementations: *com.github.castorm.kafka.connect.http.request.template.freemarker.BackwardsCompatibleFreeMarkerTemplateFactoryImplementation based on [FreeMarker](https://freemarker.apache.org/) which accepts offset properties withoutoffsetnamespace _(Deprecated)_ *com.github.castorm.kafka.connect.http.request.template.freemarker.FreeMarkerTemplateFactoryImplementation based on [FreeMarker](https://freemarker.apache.org/) *com.github.castorm.kafka.connect.http.request.template.NoTemplateFactory`

Creating a HttpRequest with FreeMarkerTemplateFactory

FreeMarker templates will have the following data model available: * offset * key * timestamp (as ISO8601 string, e.g.: 2020-01-01T00:00:00Z) * ... (custom offset properties)

Accessing any of the above withing a template can be achieved like this:

http.request.params=after=${offset.timestamp}

For an Epoch representation of the same string, FreeMarker built-ins should be used:

http.request.params=after=${offset.timestamp?datetime.iso?long}

HttpClient: Executing a HttpRequest

Once our HttpRequest is ready, we have to execute it to get some results out of it. That's the purpose of the HttpClient

http.client

```java public interface HttpClient extends Configurable {

HttpResponse execute(HttpRequest request) throws IOException;

} `` * Type:Class* Default:com.github.castorm.kafka.connect.http.client.okhttp.OkHttpClient* Available implementations: *com.github.castorm.kafka.connect.http.client.okhttp.OkHttpClient`

Executing a HttpRequest with OkHttpClient

HttpAuthenticator: Authenticating a HttpRequest

When executing the request, authentication might be required. The HttpAuthenticator is responsible for resolving the Authorization header to be included in the HttpRequest.

http.auth

```java public interface HttpAuthenticator extends Configurable {

Optional<String> getAuthorizationHeader();

} `` * Type:Class* Default:com.github.castorm.kafka.connect.http.auth.ConfigurableHttpAuthenticator* Available implementations: *com.github.castorm.kafka.connect.http.auth.ConfigurableHttpAuthenticator*com.github.castorm.kafka.connect.http.auth.NoneHttpAuthenticator*com.github.castorm.kafka.connect.http.auth.BasicHttpAuthenticator`

Authenticating with ConfigurableHttpAuthenticator

Allows selecting the authentication type via configuration property

http.auth.type

Type of authentication * Type: Enum { None, Basic } * Default: None

Authenticating with BasicHttpAuthenticator

Allows selecting the authentication type via configuration property

http.auth.user

• Type: String
• Default: ""

http.auth.password

• Type: String
• Default: """

HttpResponseParser: Parsing a HttpResponse

Once our HttpRequest has been executed, as a result we'll have to deal with a HttpResponse and translate it into the list of SourceRecords expected by Kafka Connect.

http.response.parser

```java public interface HttpResponseParser extends Configurable {

List<SourceRecord> parse(HttpResponse response);

} `` * Type:Class* Default:com.github.castorm.kafka.connect.http.response.PolicyHttpResponseParser* Available implementations: *com.github.castorm.kafka.connect.http.response.PolicyHttpResponseParser*com.github.castorm.kafka.connect.http.response.KvHttpResponseParser`

Parsing with PolicyHttpResponseParser

Vets the HTTP response deciding whether the response should be processed, skipped or failed. This decision is delegated to a HttpResponsePolicy. When the decision is to process the response, this processing is delegated to a secondary HttpResponseParser.

HttpResponsePolicy: Vetting a HttpResponse

http.response.policy

```java public interface HttpResponsePolicy extends Configurable {

HttpResponseOutcome resolve(HttpResponse response);

enum HttpResponseOutcome {
    PROCESS, SKIP, FAIL
}

} `` * Type:Class* Default:com.github.castorm.kafka.connect.http.response.StatusCodeHttpResponsePolicy* Available implementations: *com.github.castorm.kafka.connect.http.response.StatusCodeHttpResponsePolicy`

http.response.policy.parser

• Type: Class
• Default: com.github.castorm.kafka.connect.http.response.KvHttpResponseParser
• Available implementations:
  • com.github.castorm.kafka.connect.http.response.KvHttpResponseParser

Vetting with StatusCodeHttpResponsePolicy

Does response vetting based on HTTP status codes in the response and the configuration below.

http.response.policy.codes.process

Comma separated list of code ranges that will result in the parser processing the response * Example: 200..205, 207..210 * Type: String * Default: 200..299

http.response.policy.codes.skip

Comma separated list of code ranges that will result in the parser skipping the response * Example: 300..305, 307..310 * Type: String * Defa