We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
No exception when connecting to Kafka fails
Hey guys,

I have installed feast(0.4.4) via Helm3 on GKE 

The Basic example is working until the ingestion (online serving) part

There I get ->>

Waiting for feature set to be ready for ingestion...
  0%|          | 0/15 [00:00<?, ?rows/s]

but there is no progress.



When I look on GCPs bigquery interface I can see that the project "customer_project" is created with the correct columns in "customer_transactions".
But for sure no data



get_feature_set gives me 


{
  "spec": {
    "name": "customer_transactions",
    "version": 1,
    "entities": [
      {
        "name": "customer_id",
        "valueType": "INT64"
      }
    ],
    "features": [
      {
        "name": "daily_transactions",
        "valueType": "DOUBLE"
      },
      {
        "name": "total_transactions",
        "valueType": "INT64"
      }
    ],
    "maxAge": "432000s",
    "source": {
      "type": "KAFKA",
      "kafkaSourceConfig": {
        "bootstrapServers": "feast-kafka:9092",
        "topic": "feast"
      }
    },
    "project": "customer_project_1"
  },
  "meta": {
    "createdTimestamp": "2020-04-15T10:26:51Z",
    "status": "STATUS_READY"
  }
}


I had to modify some port service setup in the chart so it can be that some of feast have connection issues between kafka etc.
But There are no errors in the logs of the core and serving pod.

What can be the problem and how is a way to debug that?

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `sdk/python/feast/loaders/abstract_producer.py`
Content:
```
1 # Copyright 2019 The Feast Authors
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #     https://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 from typing import Optional, Union
16 
17 from tqdm import tqdm
18 
19 
20 class AbstractProducer:
21     """
22     Abstract class for Kafka producers
23     """
24 
25     def __init__(self, brokers: str, row_count: int, disable_progress_bar: bool):
26         self.brokers = brokers
27         self.row_count = row_count
28         self.error_count = 0
29         self.last_exception = ""
30 
31         # Progress bar will always display average rate
32         self.pbar = tqdm(
33             total=row_count, unit="rows", smoothing=0, disable=disable_progress_bar
34         )
35 
36     def produce(self, topic: str, data: bytes):
37         message = "{} should implement a produce method".format(self.__class__.__name__)
38         raise NotImplementedError(message)
39 
40     def flush(self, timeout: int):
41         message = "{} should implement a flush method".format(self.__class__.__name__)
42         raise NotImplementedError(message)
43 
44     def _inc_pbar(self, meta):
45         self.pbar.update(1)
46 
47     def _set_error(self, exception: str):
48         self.error_count += 1
49         self.last_exception = exception
50 
51     def print_results(self) -> None:
52         """
53         Print ingestion statistics.
54 
55         Returns:
56             None: None
57         """
58         # Refresh and close tqdm progress bar
59         self.pbar.refresh()
60 
61         self.pbar.close()
62 
63         print("Ingestion complete!")
64 
65         failed_message = (
66             ""
67             if self.error_count == 0
68             else f"\nFail: {self.error_count / self.row_count}"
69         )
70 
71         last_exception_message = (
72             ""
73             if self.last_exception == ""
74             else f"\nLast exception:\n{self.last_exception}"
75         )
76 
77         print(
78             f"\nIngestion statistics:"
79             f"\nSuccess: {self.pbar.n}/{self.row_count}"
80             f"{failed_message}"
81             f"{last_exception_message}"
82         )
83         return None
84 
85 
86 class ConfluentProducer(AbstractProducer):
87     """
88     Concrete implementation of Confluent Kafka producer (confluent-kafka)
89     """
90 
91     def __init__(self, brokers: str, row_count: int, disable_progress_bar: bool):
92         from confluent_kafka import Producer
93 
94         self.producer = Producer({"bootstrap.servers": brokers})
95         super().__init__(brokers, row_count, disable_progress_bar)
96 
97     def produce(self, topic: str, value: bytes) -> None:
98         """
99         Generic produce that implements confluent-kafka's produce method to
100         push a byte encoded object into a Kafka topic.
101 
102         Args:
103             topic (str): Kafka topic.
104             value (bytes): Byte encoded object.
105 
106         Returns:
107             None: None.
108         """
109 
110         try:
111             self.producer.produce(topic, value=value, callback=self._delivery_callback)
112             # Serve delivery callback queue.
113             # NOTE: Since produce() is an asynchronous API this poll() call
114             #       will most likely not serve the delivery callback for the
115             #       last produce()d message.
116             self.producer.poll(0)
117         except Exception as ex:
118             self._set_error(str(ex))
119 
120         return None
121 
122     def flush(self, timeout: Optional[int]):
123         """
124         Generic flush that implements confluent-kafka's flush method.
125 
126         Args:
127             timeout (Optional[int]): Timeout in seconds to wait for completion.
128 
129         Returns:
130             int: Number of messages still in queue.
131         """
132         return self.producer.flush(timeout=timeout)
133 
134     def _delivery_callback(self, err: str, msg) -> None:
135         """
136         Optional per-message delivery callback (triggered by poll() or flush())
137         when a message has been successfully delivered or permanently failed
138         delivery (after retries).
139 
140         Although the msg argument is not used, the current method signature is
141         required as specified in the confluent-kafka documentation.
142 
143         Args:
144             err (str): Error message.
145             msg (): Kafka message.
146 
147         Returns:
148             None
149         """
150         if err:
151             self._set_error(err)
152         else:
153             self._inc_pbar(None)
154 
155 
156 class KafkaPythonProducer(AbstractProducer):
157     """
158     Concrete implementation of Python Kafka producer (kafka-python)
159     """
160 
161     def __init__(self, brokers: str, row_count: int, disable_progress_bar: bool):
162         from kafka import KafkaProducer
163 
164         self.producer = KafkaProducer(bootstrap_servers=[brokers])
165         super().__init__(brokers, row_count, disable_progress_bar)
166 
167     def produce(self, topic: str, value: bytes):
168         """
169         Generic produce that implements kafka-python's send method to push a
170         byte encoded object into a Kafka topic.
171 
172         Args:
173             topic (str): Kafka topic.
174             value (bytes): Byte encoded object.
175 
176         Returns:
177             FutureRecordMetadata: resolves to RecordMetadata
178 
179         Raises:
180             KafkaTimeoutError: if unable to fetch topic metadata, or unable
181                 to obtain memory buffer prior to configured max_block_ms
182         """
183         return (
184             self.producer.send(topic, value=value)
185             .add_callback(self._inc_pbar)
186             .add_errback(self._set_error)
187         )
188 
189     def flush(self, timeout: Optional[int]):
190         """
191         Generic flush that implements kafka-python's flush method.
192 
193         Args:
194             timeout (Optional[int]): timeout in seconds to wait for completion.
195 
196         Returns:
197             None
198 
199         Raises:
200             KafkaTimeoutError: failure to flush buffered records within the
201                 provided timeout
202         """
203         return self.producer.flush(timeout=timeout)
204 
205 
206 def get_producer(
207     brokers: str, row_count: int, disable_progress_bar: bool
208 ) -> Union[ConfluentProducer, KafkaPythonProducer]:
209     """
210     Simple context helper function that returns a AbstractProducer object when
211     invoked.
212 
213     This helper function will try to import confluent-kafka as a producer first.
214 
215     This helper function will fallback to kafka-python if it fails to import
216     confluent-kafka.
217 
218     Args:
219         brokers (str): Kafka broker information with hostname and port.
220         row_count (int): Number of rows in table
221 
222     Returns:
223         Union[ConfluentProducer, KafkaPythonProducer]:
224             Concrete implementation of a Kafka producer. Ig can be:
225                 * confluent-kafka producer
226                 * kafka-python producer
227     """
228     try:
229         return ConfluentProducer(brokers, row_count, disable_progress_bar)
230     except ImportError:
231         print("Unable to import confluent-kafka, falling back to kafka-python")
232         return KafkaPythonProducer(brokers, row_count, disable_progress_bar)
233
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/sdk/python/feast/loaders/abstract_producer.py b/sdk/python/feast/loaders/abstract_producer.py
--- a/sdk/python/feast/loaders/abstract_producer.py
+++ b/sdk/python/feast/loaders/abstract_producer.py
@@ -25,8 +25,6 @@
     def __init__(self, brokers: str, row_count: int, disable_progress_bar: bool):
         self.brokers = brokers
         self.row_count = row_count
-        self.error_count = 0
-        self.last_exception = ""
 
         # Progress bar will always display average rate
         self.pbar = tqdm(
@@ -45,8 +43,7 @@
         self.pbar.update(1)
 
     def _set_error(self, exception: str):
-        self.error_count += 1
-        self.last_exception = exception
+        raise Exception(exception)
 
     def print_results(self) -> None:
         """
@@ -62,24 +59,7 @@
 
         print("Ingestion complete!")
 
-        failed_message = (
-            ""
-            if self.error_count == 0
-            else f"\nFail: {self.error_count / self.row_count}"
-        )
-
-        last_exception_message = (
-            ""
-            if self.last_exception == ""
-            else f"\nLast exception:\n{self.last_exception}"
-        )
-
-        print(
-            f"\nIngestion statistics:"
-            f"\nSuccess: {self.pbar.n}/{self.row_count}"
-            f"{failed_message}"
-            f"{last_exception_message}"
-        )
+        print(f"\nIngestion statistics:" f"\nSuccess: {self.pbar.n}/{self.row_count}")
         return None
 
 
@@ -129,7 +109,10 @@
         Returns:
             int: Number of messages still in queue.
         """
-        return self.producer.flush(timeout=timeout)
+        messages = self.producer.flush(timeout=timeout)
+        if messages:
+            raise Exception("Not all Kafka messages are successfully delivered.")
+        return messages
 
     def _delivery_callback(self, err: str, msg) -> None:
         """
@@ -200,7 +183,10 @@
             KafkaTimeoutError: failure to flush buffered records within the
                 provided timeout
         """
-        return self.producer.flush(timeout=timeout)
+        messages = self.producer.flush(timeout=timeout)
+        if messages:
+            raise Exception("Not all Kafka messages are successfully delivered.")
+        return messages
 
 
 def get_producer(

{"golden_diff": "diff --git a/sdk/python/feast/loaders/abstract_producer.py b/sdk/python/feast/loaders/abstract_producer.py\n--- a/sdk/python/feast/loaders/abstract_producer.py\n+++ b/sdk/python/feast/loaders/abstract_producer.py\n@@ -25,8 +25,6 @@\n     def __init__(self, brokers: str, row_count: int, disable_progress_bar: bool):\n         self.brokers = brokers\n         self.row_count = row_count\n-        self.error_count = 0\n-        self.last_exception = \"\"\n \n         # Progress bar will always display average rate\n         self.pbar = tqdm(\n@@ -45,8 +43,7 @@\n         self.pbar.update(1)\n \n     def _set_error(self, exception: str):\n-        self.error_count += 1\n-        self.last_exception = exception\n+        raise Exception(exception)\n \n     def print_results(self) -> None:\n         \"\"\"\n@@ -62,24 +59,7 @@\n \n         print(\"Ingestion complete!\")\n \n-        failed_message = (\n-            \"\"\n-            if self.error_count == 0\n-            else f\"\\nFail: {self.error_count / self.row_count}\"\n-        )\n-\n-        last_exception_message = (\n-            \"\"\n-            if self.last_exception == \"\"\n-            else f\"\\nLast exception:\\n{self.last_exception}\"\n-        )\n-\n-        print(\n-            f\"\\nIngestion statistics:\"\n-            f\"\\nSuccess: {self.pbar.n}/{self.row_count}\"\n-            f\"{failed_message}\"\n-            f\"{last_exception_message}\"\n-        )\n+        print(f\"\\nIngestion statistics:\" f\"\\nSuccess: {self.pbar.n}/{self.row_count}\")\n         return None\n \n \n@@ -129,7 +109,10 @@\n         Returns:\n             int: Number of messages still in queue.\n         \"\"\"\n-        return self.producer.flush(timeout=timeout)\n+        messages = self.producer.flush(timeout=timeout)\n+        if messages:\n+            raise Exception(\"Not all Kafka messages are successfully delivered.\")\n+        return messages\n \n     def _delivery_callback(self, err: str, msg) -> None:\n         \"\"\"\n@@ -200,7 +183,10 @@\n             KafkaTimeoutError: failure to flush buffered records within the\n                 provided timeout\n         \"\"\"\n-        return self.producer.flush(timeout=timeout)\n+        messages = self.producer.flush(timeout=timeout)\n+        if messages:\n+            raise Exception(\"Not all Kafka messages are successfully delivered.\")\n+        return messages\n \n \n def get_producer(\n", "issue": "No exception when connecting to Kafka fails\nHey guys,\r\n\r\nI have installed feast(0.4.4) via Helm3 on GKE \r\n\r\nThe Basic example is working until the ingestion (online serving) part\r\n\r\nThere I get ->>\r\n\r\nWaiting for feature set to be ready for ingestion...\r\n  0%|          | 0/15 [00:00<?, ?rows/s]\r\n\r\nbut there is no progress.\r\n\r\n\r\n\r\nWhen I look on GCPs bigquery interface I can see that the project \"customer_project\" is created with the correct columns in \"customer_transactions\".\r\nBut for sure no data\r\n\r\n\r\n\r\nget_feature_set gives me \r\n\r\n\r\n{\r\n  \"spec\": {\r\n    \"name\": \"customer_transactions\",\r\n    \"version\": 1,\r\n    \"entities\": [\r\n      {\r\n        \"name\": \"customer_id\",\r\n        \"valueType\": \"INT64\"\r\n      }\r\n    ],\r\n    \"features\": [\r\n      {\r\n        \"name\": \"daily_transactions\",\r\n        \"valueType\": \"DOUBLE\"\r\n      },\r\n      {\r\n        \"name\": \"total_transactions\",\r\n        \"valueType\": \"INT64\"\r\n      }\r\n    ],\r\n    \"maxAge\": \"432000s\",\r\n    \"source\": {\r\n      \"type\": \"KAFKA\",\r\n      \"kafkaSourceConfig\": {\r\n        \"bootstrapServers\": \"feast-kafka:9092\",\r\n        \"topic\": \"feast\"\r\n      }\r\n    },\r\n    \"project\": \"customer_project_1\"\r\n  },\r\n  \"meta\": {\r\n    \"createdTimestamp\": \"2020-04-15T10:26:51Z\",\r\n    \"status\": \"STATUS_READY\"\r\n  }\r\n}\r\n\r\n\r\nI had to modify some port service setup in the chart so it can be that some of feast have connection issues between kafka etc.\r\nBut There are no errors in the logs of the core and serving pod.\r\n\r\nWhat can be the problem and how is a way to debug that?\n", "before_files": [{"content": "# Copyright 2019 The Feast Authors\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     https://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nfrom typing import Optional, Union\n\nfrom tqdm import tqdm\n\n\nclass AbstractProducer:\n    \"\"\"\n    Abstract class for Kafka producers\n    \"\"\"\n\n    def __init__(self, brokers: str, row_count: int, disable_progress_bar: bool):\n        self.brokers = brokers\n        self.row_count = row_count\n        self.error_count = 0\n        self.last_exception = \"\"\n\n        # Progress bar will always display average rate\n        self.pbar = tqdm(\n            total=row_count, unit=\"rows\", smoothing=0, disable=disable_progress_bar\n        )\n\n    def produce(self, topic: str, data: bytes):\n        message = \"{} should implement a produce method\".format(self.__class__.__name__)\n        raise NotImplementedError(message)\n\n    def flush(self, timeout: int):\n        message = \"{} should implement a flush method\".format(self.__class__.__name__)\n        raise NotImplementedError(message)\n\n    def _inc_pbar(self, meta):\n        self.pbar.update(1)\n\n    def _set_error(self, exception: str):\n        self.error_count += 1\n        self.last_exception = exception\n\n    def print_results(self) -> None:\n        \"\"\"\n        Print ingestion statistics.\n\n        Returns:\n            None: None\n        \"\"\"\n        # Refresh and close tqdm progress bar\n        self.pbar.refresh()\n\n        self.pbar.close()\n\n        print(\"Ingestion complete!\")\n\n        failed_message = (\n            \"\"\n            if self.error_count == 0\n            else f\"\\nFail: {self.error_count / self.row_count}\"\n        )\n\n        last_exception_message = (\n            \"\"\n            if self.last_exception == \"\"\n            else f\"\\nLast exception:\\n{self.last_exception}\"\n        )\n\n        print(\n            f\"\\nIngestion statistics:\"\n            f\"\\nSuccess: {self.pbar.n}/{self.row_count}\"\n            f\"{failed_message}\"\n            f\"{last_exception_message}\"\n        )\n        return None\n\n\nclass ConfluentProducer(AbstractProducer):\n    \"\"\"\n    Concrete implementation of Confluent Kafka producer (confluent-kafka)\n    \"\"\"\n\n    def __init__(self, brokers: str, row_count: int, disable_progress_bar: bool):\n        from confluent_kafka import Producer\n\n        self.producer = Producer({\"bootstrap.servers\": brokers})\n        super().__init__(brokers, row_count, disable_progress_bar)\n\n    def produce(self, topic: str, value: bytes) -> None:\n        \"\"\"\n        Generic produce that implements confluent-kafka's produce method to\n        push a byte encoded object into a Kafka topic.\n\n        Args:\n            topic (str): Kafka topic.\n            value (bytes): Byte encoded object.\n\n        Returns:\n            None: None.\n        \"\"\"\n\n        try:\n            self.producer.produce(topic, value=value, callback=self._delivery_callback)\n            # Serve delivery callback queue.\n            # NOTE: Since produce() is an asynchronous API this poll() call\n            #       will most likely not serve the delivery callback for the\n            #       last produce()d message.\n            self.producer.poll(0)\n        except Exception as ex:\n            self._set_error(str(ex))\n\n        return None\n\n    def flush(self, timeout: Optional[int]):\n        \"\"\"\n        Generic flush that implements confluent-kafka's flush method.\n\n        Args:\n            timeout (Optional[int]): Timeout in seconds to wait for completion.\n\n        Returns:\n            int: Number of messages still in queue.\n        \"\"\"\n        return self.producer.flush(timeout=timeout)\n\n    def _delivery_callback(self, err: str, msg) -> None:\n        \"\"\"\n        Optional per-message delivery callback (triggered by poll() or flush())\n        when a message has been successfully delivered or permanently failed\n        delivery (after retries).\n\n        Although the msg argument is not used, the current method signature is\n        required as specified in the confluent-kafka documentation.\n\n        Args:\n            err (str): Error message.\n            msg (): Kafka message.\n\n        Returns:\n            None\n        \"\"\"\n        if err:\n            self._set_error(err)\n        else:\n            self._inc_pbar(None)\n\n\nclass KafkaPythonProducer(AbstractProducer):\n    \"\"\"\n    Concrete implementation of Python Kafka producer (kafka-python)\n    \"\"\"\n\n    def __init__(self, brokers: str, row_count: int, disable_progress_bar: bool):\n        from kafka import KafkaProducer\n\n        self.producer = KafkaProducer(bootstrap_servers=[brokers])\n        super().__init__(brokers, row_count, disable_progress_bar)\n\n    def produce(self, topic: str, value: bytes):\n        \"\"\"\n        Generic produce that implements kafka-python's send method to push a\n        byte encoded object into a Kafka topic.\n\n        Args:\n            topic (str): Kafka topic.\n            value (bytes): Byte encoded object.\n\n        Returns:\n            FutureRecordMetadata: resolves to RecordMetadata\n\n        Raises:\n            KafkaTimeoutError: if unable to fetch topic metadata, or unable\n                to obtain memory buffer prior to configured max_block_ms\n        \"\"\"\n        return (\n            self.producer.send(topic, value=value)\n            .add_callback(self._inc_pbar)\n            .add_errback(self._set_error)\n        )\n\n    def flush(self, timeout: Optional[int]):\n        \"\"\"\n        Generic flush that implements kafka-python's flush method.\n\n        Args:\n            timeout (Optional[int]): timeout in seconds to wait for completion.\n\n        Returns:\n            None\n\n        Raises:\n            KafkaTimeoutError: failure to flush buffered records within the\n                provided timeout\n        \"\"\"\n        return self.producer.flush(timeout=timeout)\n\n\ndef get_producer(\n    brokers: str, row_count: int, disable_progress_bar: bool\n) -> Union[ConfluentProducer, KafkaPythonProducer]:\n    \"\"\"\n    Simple context helper function that returns a AbstractProducer object when\n    invoked.\n\n    This helper function will try to import confluent-kafka as a producer first.\n\n    This helper function will fallback to kafka-python if it fails to import\n    confluent-kafka.\n\n    Args:\n        brokers (str): Kafka broker information with hostname and port.\n        row_count (int): Number of rows in table\n\n    Returns:\n        Union[ConfluentProducer, KafkaPythonProducer]:\n            Concrete implementation of a Kafka producer. Ig can be:\n                * confluent-kafka producer\n                * kafka-python producer\n    \"\"\"\n    try:\n        return ConfluentProducer(brokers, row_count, disable_progress_bar)\n    except ImportError:\n        print(\"Unable to import confluent-kafka, falling back to kafka-python\")\n        return KafkaPythonProducer(brokers, row_count, disable_progress_bar)\n", "path": "sdk/python/feast/loaders/abstract_producer.py"}], "after_files": [{"content": "# Copyright 2019 The Feast Authors\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     https://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nfrom typing import Optional, Union\n\nfrom tqdm import tqdm\n\n\nclass AbstractProducer:\n    \"\"\"\n    Abstract class for Kafka producers\n    \"\"\"\n\n    def __init__(self, brokers: str, row_count: int, disable_progress_bar: bool):\n        self.brokers = brokers\n        self.row_count = row_count\n\n        # Progress bar will always display average rate\n        self.pbar = tqdm(\n            total=row_count, unit=\"rows\", smoothing=0, disable=disable_progress_bar\n        )\n\n    def produce(self, topic: str, data: bytes):\n        message = \"{} should implement a produce method\".format(self.__class__.__name__)\n        raise NotImplementedError(message)\n\n    def flush(self, timeout: int):\n        message = \"{} should implement a flush method\".format(self.__class__.__name__)\n        raise NotImplementedError(message)\n\n    def _inc_pbar(self, meta):\n        self.pbar.update(1)\n\n    def _set_error(self, exception: str):\n        raise Exception(exception)\n\n    def print_results(self) -> None:\n        \"\"\"\n        Print ingestion statistics.\n\n        Returns:\n            None: None\n        \"\"\"\n        # Refresh and close tqdm progress bar\n        self.pbar.refresh()\n\n        self.pbar.close()\n\n        print(\"Ingestion complete!\")\n\n        print(f\"\\nIngestion statistics:\" f\"\\nSuccess: {self.pbar.n}/{self.row_count}\")\n        return None\n\n\nclass ConfluentProducer(AbstractProducer):\n    \"\"\"\n    Concrete implementation of Confluent Kafka producer (confluent-kafka)\n    \"\"\"\n\n    def __init__(self, brokers: str, row_count: int, disable_progress_bar: bool):\n        from confluent_kafka import Producer\n\n        self.producer = Producer({\"bootstrap.servers\": brokers})\n        super().__init__(brokers, row_count, disable_progress_bar)\n\n    def produce(self, topic: str, value: bytes) -> None:\n        \"\"\"\n        Generic produce that implements confluent-kafka's produce method to\n        push a byte encoded object into a Kafka topic.\n\n        Args:\n            topic (str): Kafka topic.\n            value (bytes): Byte encoded object.\n\n        Returns:\n            None: None.\n        \"\"\"\n\n        try:\n            self.producer.produce(topic, value=value, callback=self._delivery_callback)\n            # Serve delivery callback queue.\n            # NOTE: Since produce() is an asynchronous API this poll() call\n            #       will most likely not serve the delivery callback for the\n            #       last produce()d message.\n            self.producer.poll(0)\n        except Exception as ex:\n            self._set_error(str(ex))\n\n        return None\n\n    def flush(self, timeout: Optional[int]):\n        \"\"\"\n        Generic flush that implements confluent-kafka's flush method.\n\n        Args:\n            timeout (Optional[int]): Timeout in seconds to wait for completion.\n\n        Returns:\n            int: Number of messages still in queue.\n        \"\"\"\n        messages = self.producer.flush(timeout=timeout)\n        if messages:\n            raise Exception(\"Not all Kafka messages are successfully delivered.\")\n        return messages\n\n    def _delivery_callback(self, err: str, msg) -> None:\n        \"\"\"\n        Optional per-message delivery callback (triggered by poll() or flush())\n        when a message has been successfully delivered or permanently failed\n        delivery (after retries).\n\n        Although the msg argument is not used, the current method signature is\n        required as specified in the confluent-kafka documentation.\n\n        Args:\n            err (str): Error message.\n            msg (): Kafka message.\n\n        Returns:\n            None\n        \"\"\"\n        if err:\n            self._set_error(err)\n        else:\n            self._inc_pbar(None)\n\n\nclass KafkaPythonProducer(AbstractProducer):\n    \"\"\"\n    Concrete implementation of Python Kafka producer (kafka-python)\n    \"\"\"\n\n    def __init__(self, brokers: str, row_count: int, disable_progress_bar: bool):\n        from kafka import KafkaProducer\n\n        self.producer = KafkaProducer(bootstrap_servers=[brokers])\n        super().__init__(brokers, row_count, disable_progress_bar)\n\n    def produce(self, topic: str, value: bytes):\n        \"\"\"\n        Generic produce that implements kafka-python's send method to push a\n        byte encoded object into a Kafka topic.\n\n        Args:\n            topic (str): Kafka topic.\n            value (bytes): Byte encoded object.\n\n        Returns:\n            FutureRecordMetadata: resolves to RecordMetadata\n\n        Raises:\n            KafkaTimeoutError: if unable to fetch topic metadata, or unable\n                to obtain memory buffer prior to configured max_block_ms\n        \"\"\"\n        return (\n            self.producer.send(topic, value=value)\n            .add_callback(self._inc_pbar)\n            .add_errback(self._set_error)\n        )\n\n    def flush(self, timeout: Optional[int]):\n        \"\"\"\n        Generic flush that implements kafka-python's flush method.\n\n        Args:\n            timeout (Optional[int]): timeout in seconds to wait for completion.\n\n        Returns:\n            None\n\n        Raises:\n            KafkaTimeoutError: failure to flush buffered records within the\n                provided timeout\n        \"\"\"\n        messages = self.producer.flush(timeout=timeout)\n        if messages:\n            raise Exception(\"Not all Kafka messages are successfully delivered.\")\n        return messages\n\n\ndef get_producer(\n    brokers: str, row_count: int, disable_progress_bar: bool\n) -> Union[ConfluentProducer, KafkaPythonProducer]:\n    \"\"\"\n    Simple context helper function that returns a AbstractProducer object when\n    invoked.\n\n    This helper function will try to import confluent-kafka as a producer first.\n\n    This helper function will fallback to kafka-python if it fails to import\n    confluent-kafka.\n\n    Args:\n        brokers (str): Kafka broker information with hostname and port.\n        row_count (int): Number of rows in table\n\n    Returns:\n        Union[ConfluentProducer, KafkaPythonProducer]:\n            Concrete implementation of a Kafka producer. Ig can be:\n                * confluent-kafka producer\n                * kafka-python producer\n    \"\"\"\n    try:\n        return ConfluentProducer(brokers, row_count, disable_progress_bar)\n    except ImportError:\n        print(\"Unable to import confluent-kafka, falling back to kafka-python\")\n        return KafkaPythonProducer(brokers, row_count, disable_progress_bar)\n", "path": "sdk/python/feast/loaders/abstract_producer.py"}]}