Source Kafka: added option for giving polling time when consuming records using consumer (#12903)

* added option for giving polling time using spec

* add doc and seed file

Co-authored-by: marcosmarxm <[email protected]>

Author: tirth7777777

airbyte-config/init/src/main/resources/seed/source_definitions.yaml

@@ -446,7 +446,7 @@
 - name: Kafka
   sourceDefinitionId: d917a47b-8537-4d0d-8c10-36a9928d4265
   dockerRepository: airbyte/source-kafka
-  dockerImageTag: 0.1.5
+  dockerImageTag: 0.1.6
   documentationUrl: https://docs.airbyte.io/integrations/sources/kafka
   icon: kafka.svg
   sourceType: database

airbyte-config/init/src/main/resources/seed/source_specs.yaml

@@ -3969,7 +3969,7 @@
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
-- dockerImage: "airbyte/source-kafka:0.1.5"
+- dockerImage: "airbyte/source-kafka:0.1.6"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/sources/kafka"
     connectionSpecification:

airbyte-integrations/connectors/source-kafka/Dockerfile

@@ -16,5 +16,5 @@ ENV APPLICATION source-kafka
 
 COPY --from=build /airbyte /airbyte
 
-LABEL io.airbyte.version=0.1.5
+LABEL io.airbyte.version=0.1.6
 LABEL io.airbyte.name=airbyte/source-kafka

airbyte-integrations/connectors/source-kafka/src/main/java/io/airbyte/integrations/source/kafka/KafkaSource.java

@@ -90,8 +90,9 @@ public AutoCloseableIterator<AirbyteMessage> read(final JsonNode config, final C
 
     final int retry = config.has("repeated_calls") ? config.get("repeated_calls").intValue() : 0;
     int pollCount = 0;
+    final int polling_time = config.has("polling_time") ? config.get("polling_time").intValue() : 100;
     while (true) {
-      final ConsumerRecords<String, JsonNode> consumerRecords = consumer.poll(Duration.of(100, ChronoUnit.MILLIS));
+      final ConsumerRecords<String, JsonNode> consumerRecords = consumer.poll(Duration.of(polling_time, ChronoUnit.MILLIS));
       if (consumerRecords.count() == 0) {
         pollCount++;
         if (pollCount > retry) {

airbyte-integrations/connectors/source-kafka/src/main/resources/spec.json

@@ -78,6 +78,12 @@
         "type": "integer",
         "default": 500
       },
+      "polling_time": {
+        "title": "Polling Time",
+        "description": "Amount of time Kafka connector should try to poll for messages.",
+        "type": "integer",
+        "default": 100
+      },
       "protocol": {
         "title": "Protocol",
         "type": "object",

docs/integrations/sources/kafka.md

@@ -44,6 +44,7 @@ The Kafka source connector supports the following[sync modes](https://docs.airby
 
 | Version | Date       | Pull Request                                           | Subject                                   |
 | :------ | :--------  | :------------------------------------------------------| :---------------------------------------- |
+| 0.1.6   | 2022-05-29 | [12903](https://github.com/airbytehq/airbyte/pull/12903) | Add Polling Time to Specification (default 100 ms) |
 | 0.1.5   | 2022-04-19 | [12134](https://github.com/airbytehq/airbyte/pull/12134) | Add PLAIN Auth |
 | 0.1.4   | 2022-02-15 | [10186](https://github.com/airbytehq/airbyte/pull/10186) | Add SCRAM-SHA-512 Auth |
 | 0.1.3   | 2022-02-14 | [10256](https://github.com/airbytehq/airbyte/pull/10256) | Add `-XX:+ExitOnOutOfMemoryError` JVM option |