Fix bare except

confluent_kafka/avro/cached_schema_registry_client.py

@@ -168,9 +168,9 @@ def get_by_id(self, schema_id):
                 # cache it
                 self._cache_schema(result, schema_id)
                 return result
-            except:
+            except ClientError as e:
                 # bad schema - should not happen
-                raise ClientError("Received bad schema from registry.")
+                raise ClientError("Received bad schema (id %s) from registry: %s" % (schema_id, e))
 
     def get_latest_schema(self, subject):
         """
@@ -204,9 +204,9 @@ def get_latest_schema(self, subject):
         else:
             try:
                 schema = loads(result['schema'])
-            except:
+            except ClientError:
                 # bad schema - should not happen
-                raise ClientError("Received bad schema from registry.")
+                raise
 
         self._cache_schema(schema, schema_id, subject, version)
         return (schema_id, schema, version)
@@ -269,7 +269,8 @@ def test_compatibility(self, subject, avro_schema, version='latest'):
             else:
                 log.error("Unable to check the compatibility")
                 False
-        except:
+        except Exception as e:
+            log.error("_send_request() failed: %s", e)
             return False
 
     def update_compatibility(self, level, subject=None):

confluent_kafka/avro/load.py

@@ -17,13 +17,18 @@
 
 import sys
 
+from confluent_kafka.avro.error import ClientError
+
 
 def loads(schema_str):
     """ Parse a schema given a schema string """
-    if sys.version_info[0] < 3:
-        return schema.parse(schema_str)
-    else:
-        return schema.Parse(schema_str)
+    try:
+        if sys.version_info[0] < 3:
+            return schema.parse(schema_str)
+        else:
+            return schema.Parse(schema_str)
+    except schema.AvroException.SchemaParseException as e:
+        raise ClientError("Schema parse failed: %s" % (str(e)))
 
 
 def load(fp):
@@ -44,5 +49,6 @@ def _hash_func(self):
     schema.RecordSchema.__hash__ = _hash_func
     schema.PrimitiveSchema.__hash__ = _hash_func
     schema.UnionSchema.__hash__ = _hash_func
+
 except ImportError:
     schema = None

confluent_kafka/avro/serializer/message_serializer.py

@@ -42,7 +42,7 @@
     from fastavro.reader import read_data
 
     HAS_FAST = True
-except:
+except ImportError:
     pass
 
 
@@ -158,12 +158,11 @@ def _get_decoder_func(self, schema_id, payload):
         # fetch from schema reg
         try:
             schema = self.registry_client.get_by_id(schema_id)
-        except:
-            schema = None
+        except ClientError as e:
+            raise SerializerError("unable to fetch schema with id %d: %s" % (schema_id, str(e)))
 
-        if not schema:
-            err = "unable to fetch schema with id %d" % (schema_id)
-            raise SerializerError(err)
+        if schema is None:
+            raise SerializerError("unable to fetch schema with id %d" % (schema_id))
 
         curr_pos = payload.tell()
         if HAS_FAST:
@@ -180,7 +179,8 @@ def _get_decoder_func(self, schema_id, payload):
 
                 self.id_to_decoder_func[schema_id] = lambda p: read_data(p, schema_dict)
                 return self.id_to_decoder_func[schema_id]
-            except:
+            except Exception:
+                # Fast avro failed, fall thru to standard avro below.
                 pass
 
         # here means we should just delegate to slow avro

examples/consumer.py

@@ -59,7 +59,7 @@ def print_usage_and_exit(program_name):
             continue
         try:
             intval = int(opt[1])
-        except:
+        except ValueError:
             sys.stderr.write("Invalid option value for -T: %s\n" % opt[1])
             sys.exit(1)
 

tests/avro/mock_registry.py

@@ -24,10 +24,12 @@
 import json
 import re
 
-from threading import Thread
+from threading import Thread, Event
 
 from tests.avro.mock_schema_registry_client import MockSchemaRegistryClient
 from confluent_kafka import avro
+from confluent_kafka.avro.error import ClientError
+
 
 if sys.version_info[0] < 3:
     import BaseHTTPServer as HTTPSERVER
@@ -120,7 +122,7 @@ def _get_schema_from_body(self, req):
         try:
             avro_schema = avro.loads(schema)
             return self._get_identity_schema(avro_schema)
-        except:
+        except ClientError:
             return None
 
     def register(self, req, groups):
@@ -174,11 +176,18 @@ def __init__(self, port):
         self.server = None
         self.port = port
         self.daemon = True
+        self.started = Event()
 
     def run(self):
         self.server = MockServer(('127.0.0.1', self.port), ReqHandler)
+        self.started.set()
         self.server.serve_forever()
 
+    def start(self):
+        """Start, and wait for server to be fully started, before returning."""
+        super(ServerThread, self).start()
+        self.started.wait()
+
     def shutdown(self):
         if self.server:
             self.server.shutdown()

tests/avro/test_cached_client.py

@@ -20,8 +20,6 @@
 # derived from https://github.com/verisign/python-confluent-schemaregistry.git
 #
 
-import time
-
 import unittest
 
 from tests.avro import mock_registry
@@ -34,7 +32,6 @@ class TestCacheSchemaRegistryClient(unittest.TestCase):
     def setUp(self):
         self.server = mock_registry.ServerThread(0)
         self.server.start()
-        time.sleep(1)
         self.client = CachedSchemaRegistryClient('http://127.0.0.1:' + str(self.server.server.server_port))
 
     def tearDown(self):

tests/test_threads.py

@@ -5,7 +5,7 @@
 import time
 try:
     from queue import Queue, Empty
-except:
+except ImportError:
     from Queue import Queue, Empty