Failed to poll records

[TartanLeGrand]

Hello,

I'm trying to use the connector to retrieve data from an api. Unfortunately the connector throws me an WARN:

kafka-connect      | [2021-12-02 10:09:30,550] WARN [test|task-0] WorkerSourceTask{id=test-0} failed to poll records from SourceTask. Will retry operation. (org.apache.kafka.connect.runtime.WorkerSourceTask:293)
kafka-connect      | org.apache.kafka.connect.errors.RetriableException: java.net.ProtocolException: Too many follow-up requests: 21
kafka-connect      | 	at com.github.castorm.kafka.connect.http.HttpSourceTask.execute(HttpSourceTask.java:128)
kafka-connect      | 	at com.github.castorm.kafka.connect.http.HttpSourceTask.poll(HttpSourceTask.java:105)
kafka-connect      | 	at org.apache.kafka.connect.runtime.WorkerSourceTask.poll(WorkerSourceTask.java:291)
kafka-connect      | 	at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:248)
kafka-connect      | 	at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:186)
kafka-connect      | 	at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:241)
kafka-connect      | 	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
kafka-connect      | 	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
kafka-connect      | 	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
kafka-connect      | 	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
kafka-connect      | 	at java.base/java.lang.Thread.run(Thread.java:829)
kafka-connect      | Caused by: java.net.ProtocolException: Too many follow-up requests: 21
kafka-connect      | 	at okhttp3.internal.http.RetryAndFollowUpInterceptor.intercept(RetryAndFollowUpInterceptor.kt:127)
kafka-connect      | 	at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.kt:109)
kafka-connect      | 	at com.github.castorm.kafka.connect.http.client.okhttp.OkHttpClient.lambda$configure$0(OkHttpClient.java:70)
kafka-connect      | 	at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.kt:109)
kafka-connect      | 	at okhttp3.logging.HttpLoggingInterceptor.intercept(HttpLoggingInterceptor.kt:154)
kafka-connect      | 	at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.kt:109)
kafka-connect      | 	at okhttp3.internal.connection.RealCall.getResponseWithInterceptorChain$okhttp(RealCall.kt:201)
kafka-connect      | 	at okhttp3.internal.connection.RealCall.execute(RealCall.kt:154)
kafka-connect      | 	at com.github.castorm.kafka.connect.http.client.okhttp.OkHttpClient.execute(OkHttpClient.java:99)
kafka-connect      | 	at com.github.castorm.kafka.connect.http.HttpSourceTask.execute(HttpSourceTask.java:126)
kafka-connect      | 	... 10 more

and ERROR:

kafka-connect      | [2021-12-02 10:35:45,355] ERROR [test|task-0|offsets] WorkerSourceTask{id=test-0} Exception thrown while calling task.commit() (org.apache.kafka.connect.runtime.WorkerSourceTask:598)
kafka-connect      | java.lang.NullPointerException
kafka-connect      | 	at com.github.castorm.kafka.connect.http.HttpSourceTask.commit(HttpSourceTask.java:139)
kafka-connect      | 	at org.apache.kafka.connect.runtime.WorkerSourceTask.commitSourceTask(WorkerSourceTask.java:596)
kafka-connect      | 	at org.apache.kafka.connect.runtime.WorkerSourceTask.commitOffsets(WorkerSourceTask.java:540)
kafka-connect      | 	at org.apache.kafka.connect.runtime.SourceTaskOffsetCommitter.commit(SourceTaskOffsetCommitter.java:110)
kafka-connect      | 	at org.apache.kafka.connect.runtime.SourceTaskOffsetCommitter.lambda$schedule$0(SourceTaskOffsetCommitter.java:84)
kafka-connect      | 	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
kafka-connect      | 	at java.base/java.util.concurrent.FutureTask.runAndReset(FutureTask.java:305)
kafka-connect      | 	at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:305)
kafka-connect      | 	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
kafka-connect      | 	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
kafka-connect      | 	at java.base/java.lang.Thread.run(Thread.java:829)

My config for the connector:

{
  "name": "test",
  "config": {
      "connector.class": "com.github.castorm.kafka.connect.http.HttpSourceConnector",
      "tasks.max": "1",
      "http.request.url": "https://api.url/",
      "http.request.method": "POST",
      "http.request.headers": "Content-Type: text/plain, Client-ID: XX', Authorization: Bearer XX",
      "http.request.body": "fields *;limit 1;",
      "http.timer.interval.millis": "10000",
      "http.timer.catchup.interval.millis": "1000",
      "kafka.topic": "topic"
  }
}

All my other tests on other api work. I don't know how to solve it. I guess you have already faced this problem.

Thanks

[lzuchowska]

I'm facing the same issue, did you perhaps manage to solve it? 🙏

[TartanLeGrand]

Hey @lzuchowska ✋

I have reduce the size of the request http. I have limited the request to 10 values returned by the API. For my last tests this worked.

[lzuchowska]

Thanks a lot! It helped me too ✨