diff --git a/kafka/producer/kafka.py b/kafka/producer/kafka.py
index 7878c0a57..d5c1c34c0 100644
--- a/kafka/producer/kafka.py
+++ b/kafka/producer/kafka.py
@@ -546,9 +546,10 @@ def send(self, topic, value=None, key=None, headers=None, partition=None, timest
             KafkaTimeoutError: if unable to fetch topic metadata, or unable
                 to obtain memory buffer prior to configured max_block_ms
         """
-        assert value is not None or self.config['api_version'] >= (0, 8, 1), (
-            'Null messages require kafka >= 0.8.1')
-        assert not (value is None and key is None), 'Need at least one: key or value'
+        if value is None and self.config['api_version'] < (0, 8, 1):
+            raise ValueError('Null messages require kafka >= 0.8.1')
+        if value is None and key is None:
+            raise ValueError('Need at least one: key or value')
         key_bytes = value_bytes = None
         try:
             self._wait_on_metadata(topic, self.config['max_block_ms'] / 1000.0)
@@ -559,18 +560,28 @@ def send(self, topic, value=None, key=None, headers=None, partition=None, timest
             value_bytes = self._serialize(
                 self.config['value_serializer'],
                 topic, value)
-            assert type(key_bytes) in (bytes, bytearray, memoryview, type(None))
-            assert type(value_bytes) in (bytes, bytearray, memoryview, type(None))
+            valid_types = (bytes, bytearray, memoryview, type(None))
+            if type(key_bytes) not in valid_types:
+                raise TypeError('key_serializer must return bytes-like data or None')
+            if type(value_bytes) not in valid_types:
+                raise TypeError('value_serializer must return bytes-like data or None')
 
             partition = self._partition(topic, partition, key, value,
                                         key_bytes, value_bytes)
-
             if headers is None:
                 headers = []
-            assert type(headers) == list
-            assert all(type(item) == tuple and len(item) == 2 and type(item[0]) == str and type(item[1]) == bytes for item in headers)
-
-            message_size = self._estimate_size_in_bytes(key_bytes, value_bytes, headers)
+            if type(headers) is not list:
+                raise TypeError('headers must be a list of (str, bytes) tuples')
+            if not all(
+                type(item) is tuple
+                and len(item) == 2
+                and type(item[0]) is str
+                and type(item[1]) is bytes
+                for item in headers
+            ):
+                raise TypeError('headers must be a list of (str, bytes) tuples')
+
+                message_size = self._estimate_size_in_bytes(key_bytes, value_bytes, headers)
             self._ensure_valid_record_size(message_size)
 
             tp = TopicPartition(topic, partition)
@@ -694,8 +705,10 @@ def _serialize(self, f, topic, data):
     def _partition(self, topic, partition, key, value,
                    serialized_key, serialized_value):
         if partition is not None:
-            assert partition >= 0
-            assert partition in self._metadata.partitions_for_topic(topic), 'Unrecognized partition'
+            if partition < 0:
+                raise ValueError('partition must be non-negative')
+            if partition not in self._metadata.partitions_for_topic(topic):
+                raise ValueError('Unrecognized partition')
             return partition
 
         all_partitions = sorted(self._metadata.partitions_for_topic(topic))