[GitHub] [airflow] luup2k commented on a change in pull request #1415: Add KafkaConsumerHook and KafkaSensor

[GitBox <gi...@apache.org>]

luup2k commented on a change in pull request #1415:
URL: https://github.com/apache/airflow/pull/1415#discussion_r579772221



##########
File path: airflow/contrib/hooks/kafka_hook.py
##########
@@ -0,0 +1,69 @@
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from airflow.hooks import BaseHook
+from kafka import KafkaConsumer
+
+
+class KafkaConsumerHook(BaseHook):
+
+    default_host = 'localhost'
+    default_port = 9092
+
+    def __init__(self, conn_id, topic):
+        super(KafkaConsumerHook, self).__init__(None)
+        self.conn = self.get_connection(conn_id)
+        self.server = None
+        self.consumer = None
+        self.topic = topic
+
+    def get_conn(self):
+        conf = self.conn.extra_dejson
+        host = self.conn.host or self.default_host
+        port = self.conn.port or self.default_port
+
+        # Disable auto commit as the hook will commit right
+        # after polling.
+        conf['enable_auto_commit'] = False
+
+        self.server = '{host}:{port}'.format(**locals())
+        self.consumer = KafkaConsumer(
+            self.topic,
+            bootstrap_servers=self.server, **conf)
+
+        return self.consumer
+
+    def get_messages(self):
+        """
+        Get all the messages haven't been consumed, it doesn't

Review comment:
       > "Get all the messages haven't been consumed,"
   
   If we use poll() without max_records, the behavior is returns at most "max_poll_records" #records. "max_poll_records" is setted to 500 by default at Consumer Init config.
   So, we're not going to consume "all" message except we put a very high number as max_poll_records(could be a memory bomb) or  we have a low number of message in the topic.
   
   https://kafka-python.readthedocs.io/en/master/apidoc/KafkaConsumer.html#kafka.KafkaConsumer.poll




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org