diff --git a/.travis.yml b/.travis.yml
index 775eca2..1ac4c18 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -4,8 +4,7 @@ services:
   - docker
 
 before_install:
-- docker pull johnnypark/kafka-zookeeper
-- docker run -td -p 2181:2181 -p 9092:9092 -e ADVERTISED_HOST=127.0.0.1 -e NUM_PARTITIONS=10 johnnypark/kafka-zookeeper
+- cd examples && docker-compose up -d && cd ..
 - docker ps -a
 - pip install -r requirements.txt
 - pip install requests
@@ -14,4 +13,4 @@ before_install:
 - pip install .
 
 script:
-- python3 -m robot examples/test.robot
+- python3 -m robot examples/
diff --git a/README.md b/README.md
index d8f3b4b..5ed2960 100644
--- a/README.md
+++ b/README.md
@@ -4,23 +4,7 @@
 
 ConfluentKafkaLibrary library is wrapper for [confluent-kafka-python](https://github.com/confluentinc/confluent-kafka-python).
 
-Still in development, right now supports:
-
-* [ ] Consumer
-  * [X] Poll
-  * [X] Un/Subscribe
-  * [X] Create / Stop consumer
-  * [X] Assign
-  * [X] List topics
-  * [ ] commit
-  * [ ] offsets
-  * [X] Run in thread
-  * [X] Decode option of data from topic
-* [X] Producer
-
-
-ConfluentKafkaLibrary works with latest confluent-kafka-python 1.0.0.
-
+ConfluentKafkaLibrary works with latest confluent-kafka-python, tags are 1:1 (ConfluentKafkaLibrary 1.3.0 == confluent-kafka-python 1.3.0 ). Bugfixes and updates are set after the '-' e.g. `1.3.0-1`.
 
 ## Documentation
 
diff --git a/docs/index.html b/docs/index.html
index 1c44673..a271733 100644
--- a/docs/index.html
+++ b/docs/index.html
@@ -5,7 +5,7 @@
 <meta http-equiv="Pragma" content="no-cache">
 <meta http-equiv="Expires" content="-1">
 <meta http-equiv="X-UA-Compatible" content="IE=edge">
-<meta content="Robot Framework 3.1.1 (Python 3.6.8 on linux)" name="Generator">
+<meta content="Robot Framework 3.1.2 (Python 3.6.8 on linux)" name="Generator">
 <link rel="icon" type="image/x-icon" href="data:image/x-icon;base64,AAABAAEAEBAAAAEAIABoBAAAFgAAACgAAAAQAAAAIAAAAAEAIAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAKcAAAD/AAAA/wAAAP8AAAD/AAAA/wAAAP8AAAD/AAAA/wAAAP8AAAD/AAAAqAAAAAAAAAAAAAAAAAAAALIAAAD/AAAA4AAAANwAAADcAAAA3AAAANwAAADcAAAA3AAAANwAAADcAAAA4AAAAP8AAACxAAAAAAAAAKYAAAD/AAAAuwAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAC/AAAA/wAAAKkAAAD6AAAAzAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAN8AAAD/AAAA+gAAAMMAAAAAAAAAAgAAAGsAAABrAAAAawAAAGsAAABrAAAAawAAAGsAAABrAAAADAAAAAAAAADaAAAA/wAAAPoAAADDAAAAAAAAAIsAAAD/AAAA/wAAAP8AAAD/AAAA/wAAAP8AAAD/AAAA/wAAANEAAAAAAAAA2gAAAP8AAAD6AAAAwwAAAAAAAAAAAAAAMgAAADIAAAAyAAAAMgAAADIAAAAyAAAAMgAAADIAAAAFAAAAAAAAANoAAAD/AAAA+gAAAMMAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAADaAAAA/wAAAPoAAADDAAAAAAAAADwAAAB8AAAAAAAAAGAAAABcAAAAAAAAAH8AAABKAAAAAAAAAAAAAAAAAAAA2gAAAP8AAAD6AAAAwwAAAAAAAADCAAAA/wAAACkAAADqAAAA4QAAAAAAAAD7AAAA/wAAALAAAAAGAAAAAAAAANoAAAD/AAAA+gAAAMMAAAAAAAAAIwAAAP4AAAD/AAAA/wAAAGAAAAAAAAAAAAAAAMkAAAD/AAAAigAAAAAAAADaAAAA/wAAAPoAAADDAAAAAAAAAAAAAAAIAAAAcAAAABkAAAAAAAAAAAAAAAAAAAAAAAAAEgAAAAAAAAAAAAAA2gAAAP8AAAD7AAAAywAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAN4AAAD/AAAAqwAAAP8AAACvAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAALIAAAD/AAAAsgAAAAAAAAC5AAAA/wAAAMoAAADAAAAAwAAAAMAAAADAAAAAwAAAAMAAAADAAAAAwAAAAMkAAAD/AAAAvAAAAAAAAAAAAAAAAAAAAKwAAAD/AAAA/wAAAP8AAAD/AAAA/wAAAP8AAAD/AAAA/wAAAP8AAAD/AAAArQAAAAAAAAAAwAMAAIABAAAf+AAAP/wAAD/8AAAgBAAAP/wAAD/8AAA//AAAJIwAADHEAAA//AAAP/wAAB/4AACAAQAAwAMAAA==">
 <style media="all" type="text/css">
 body {
@@ -547,7 +547,7 @@
 jQuery.extend({highlight:function(e,t,n,r){if(e.nodeType===3){var i=e.data.match(t);if(i){var s=document.createElement(n||"span");s.className=r||"highlight";var o=e.splitText(i.index);o.splitText(i[0].length);var u=o.cloneNode(true);s.appendChild(u);o.parentNode.replaceChild(s,o);return 1}}else if(e.nodeType===1&&e.childNodes&&!/(script|style)/i.test(e.tagName)&&!(e.tagName===n.toUpperCase()&&e.className===r)){for(var a=0;a<e.childNodes.length;a++){a+=jQuery.highlight(e.childNodes[a],t,n,r)}}return 0}});jQuery.fn.unhighlight=function(e){var t={className:"highlight",element:"span"};jQuery.extend(t,e);return this.find(t.element+"."+t.className).each(function(){var e=this.parentNode;e.replaceChild(this.firstChild,this);e.normalize()}).end()};jQuery.fn.highlight=function(e,t){var n={className:"highlight",element:"span",caseSensitive:false,wordsOnly:false};jQuery.extend(n,t);if(e.constructor===String){e=[e]}e=jQuery.grep(e,function(e,t){return e!=""});e=jQuery.map(e,function(e,t){return e.replace(/[-[\]{}()*+?.,\\^$|#\s]/g,"\\$&")});if(e.length==0){return this}var r=n.caseSensitive?"":"i";var i="("+e.join("|")+")";if(n.wordsOnly){i="\\b"+i+"\\b"}var s=new RegExp(i,r);return this.each(function(){jQuery.highlight(this,s,n.element,n.className)})}
 </script>
 <script type="text/javascript">
-libdoc = {"all_tags":[],"contains_tags":false,"doc":"<p>ConfluentKafkaLibrary is a Robot Framework library which wraps up <a href=\"https://github.com/confluentinc/confluent-kafka-python\">confluent-kafka-python\x3c/a>. Keywords are inspired by <a href=\"https://github.com/s4int/robotframework-KafkaLibrary\">robotframework-KafkaLibrary\x3c/a> which is based on <a href=\"https://github.com/Parsely/pykafka\">pykafka project\x3c/a> and support more functionality like running more clients based on <span class=\"name\">group_id\x3c/span> or running them in threaded mode during the tests, decoding of gathered data etc. (<span class=\"name\">See\x3c/span> <a href=\"#Examples\" class=\"name\">Examples\x3c/a>).\x3c/p>\n<p>This document explains how to use keywords provided by ConfluentKafkaLibrary. For information about installation, support, and more, please visit the <a href=\"https://github.com/robooo/robotframework-ConfluentKafkaLibrary\">project github page\x3c/a>. For more information about Robot Framework, see <a href=\"http://robotframework.org\">http://robotframework.org\x3c/a>.\x3c/p>\n<h3 id=\"Examples\">Examples\x3c/h3>\n<p><b>Basic Consumer with predefined group_id\x3c/b>\x3c/p>\n<table border=\"1\">\n<tr>\n<td>${group_id}=\x3c/td>\n<td><a href=\"#Create%20Consumer\" class=\"name\">Create Consumer\x3c/a>\x3c/td>\n<td>group_id=mygroup\x3c/td>\n<td># if group_id is not defined uuid4() is gemerated\x3c/td>\n\x3c/tr>\n<tr>\n<td><a href=\"#Subscribe%20Topic\" class=\"name\">Subscribe Topic\x3c/a>\x3c/td>\n<td>group_id=${group_id}\x3c/td>\n<td>topics=test_topic\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n<tr>\n<td>${result}=\x3c/td>\n<td><a href=\"#Poll\" class=\"name\">Poll\x3c/a>\x3c/td>\n<td>group_id=${group_id}\x3c/td>\n<td>max_records=5\x3c/td>\n\x3c/tr>\n<tr>\n<td><span class=\"name\">Log\x3c/span>\x3c/td>\n<td>${result}\x3c/td>\n<td>\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n<tr>\n<td><a href=\"#Unsubscribe\" class=\"name\">Unsubscribe\x3c/a>\x3c/td>\n<td>${group_id}\x3c/td>\n<td>\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n<tr>\n<td><a href=\"#Close%20Consumer\" class=\"name\">Close Consumer\x3c/a>\x3c/td>\n<td>${group_id}\x3c/td>\n<td>\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n\x3c/table>\n<p><b>More Consumers\x3c/b>\x3c/p>\n<table border=\"1\">\n<tr>\n<td>${group_id_1}=\x3c/td>\n<td><a href=\"#Create%20Consumer\" class=\"name\">Create Consumer\x3c/a>\x3c/td>\n<td>\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n<tr>\n<td><a href=\"#Subscribe%20Topic\" class=\"name\">Subscribe Topic\x3c/a>\x3c/td>\n<td>group_id=${group_id_1}\x3c/td>\n<td>topics=topic1\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n<tr>\n<td>${group_id_2}=\x3c/td>\n<td><a href=\"#Create%20Consumer\" class=\"name\">Create Consumer\x3c/a>\x3c/td>\n<td>\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n<tr>\n<td><a href=\"#Subscribe%20Topic\" class=\"name\">Subscribe Topic\x3c/a>\x3c/td>\n<td>group_id=${group_id_2}\x3c/td>\n<td>topics=topic2\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n<tr>\n<td>${result_1}=\x3c/td>\n<td><a href=\"#Poll\" class=\"name\">Poll\x3c/a>\x3c/td>\n<td>group_id=${group_id_1}\x3c/td>\n<td>max_records=5\x3c/td>\n\x3c/tr>\n<tr>\n<td>${result_2}=\x3c/td>\n<td><a href=\"#Poll\" class=\"name\">Poll\x3c/a>\x3c/td>\n<td>group_id=${group_id_2}\x3c/td>\n<td>max_records=2\x3c/td>\n\x3c/tr>\n<tr>\n<td><a href=\"#Unsubscribe\" class=\"name\">Unsubscribe\x3c/a>\x3c/td>\n<td>${group_id_1}\x3c/td>\n<td>\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n<tr>\n<td><a href=\"#Unsubscribe\" class=\"name\">Unsubscribe\x3c/a>\x3c/td>\n<td>${group_id_2}\x3c/td>\n<td>\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n<tr>\n<td><a href=\"#Close%20Consumer\" class=\"name\">Close Consumer\x3c/a>\x3c/td>\n<td>${group_id_1}\x3c/td>\n<td>\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n<tr>\n<td><a href=\"#Close%20Consumer\" class=\"name\">Close Consumer\x3c/a>\x3c/td>\n<td>${group_id_2}\x3c/td>\n<td>\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n\x3c/table>\n<p><b>Run Avro Consumer over HTTPS and threaded:\x3c/b>\x3c/p>\n<table border=\"1\">\n<tr>\n<td>${thread}=\x3c/td>\n<td><a href=\"#Start%20Consumer%20Threaded\" class=\"name\">Start Consumer Threaded\x3c/a>\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n<tr>\n<td>\x3c/td>\n<td>...  topics=test_topic\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n<tr>\n<td>\x3c/td>\n<td>...  schema_registry_url=https://localhost:8081\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n<tr>\n<td>\x3c/td>\n<td>...  auto_offset_reset=earliest\x3c/td>\n<td># We want to get all messages\x3c/td>\n\x3c/tr>\n<tr>\n<td>\x3c/td>\n<td>...  security.protocol=ssl\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n<tr>\n<td>\x3c/td>\n<td>...  schema.registry.ssl.ca.location=/home/user/cert/caproxy.pem\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n<tr>\n<td>\x3c/td>\n<td>...  ssl.ca.location=/home/user/cert/caproxy.pem\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n<tr>\n<td>\x3c/td>\n<td>...  ssl.certificate.location=/home/user/cert/kafka-client-cert.pem\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n<tr>\n<td>\x3c/td>\n<td>...  ssl.key.location=/home/user/cert/kafka-client-key.pem\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n<tr>\n<td><span class=\"name\">Log\x3c/span>\x3c/td>\n<td><span class=\"name\">Execute commands which should push some data to topic\x3c/span>\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n<tr>\n<td>${messages}=\x3c/td>\n<td><a href=\"#Get%20Messages%20From%20Thread\" class=\"name\">Get Messages From Thread\x3c/a>\x3c/td>\n<td>${thread}\x3c/td>\n\x3c/tr>\n<tr>\n<td>Stop Thread\x3c/td>\n<td>${thread}\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n\x3c/table>","generated":"2019-06-23 21:17:01","inits":[],"keywords":[{"args":["group_id","topic_partition"],"doc":"<p>Assign a list of TopicPartitions.\x3c/p>\n<ul>\n<li><code>partitions\x3c/code> (list of <span class=\"name\">TopicPartition\x3c/span>): Assignment for this instance.\x3c/li>\n\x3c/ul>","matched":true,"name":"Assign To Topic Partition","shortdoc":"Assign a list of TopicPartitions.","tags":[]},{"args":["group_id"],"doc":"<p>Close down and terminate the Kafka Consumer.\x3c/p>","matched":true,"name":"Close Consumer","shortdoc":"Close down and terminate the Kafka Consumer.","tags":[]},{"args":["group_id=None","server=127.0.0.1","port=9092","client_id=Robot","enable_auto_commit=True","auto_offset_reset=latest","schema_registry_url=None","**kwargs"],"doc":"<p>Create Kafka Consumer and returns its <span class=\"name\">group_id\x3c/span> as string.\x3c/p>\n<p>Keyword Arguments:\x3c/p>\n<ul>\n<li><code>server\x3c/code>: (str): IP address / domain, that the consumer should contact to bootstrap initial cluster metadata. Default: <span class=\"name\">127.0.0.1\x3c/span>.\x3c/li>\n<li><code>port\x3c/code> (int): Port number. Default: <span class=\"name\">9092\x3c/span>.\x3c/li>\n<li><code>client_id\x3c/code> (str): a name for this client. This string is passed in each request to servers and can be used to identify specific server-side log entries that correspond to this client. Also submitted to GroupCoordinator for logging with respect to consumer group administration. Default: <span class=\"name\">Robot\x3c/span>.\x3c/li>\n<li><code>group_id\x3c/code> (str or uuid.uuid4() if not set) : name of the consumer group to join for dynamic partition assignment (if enabled), and to use for fetching and committing offsets. If None, unique string is generated  (via uuid.uuid4()) and offset commits are disabled. Default: <span class=\"name\">None\x3c/span>.\x3c/li>\n<li><code>auto_offset_reset\x3c/code> (str): A policy for resetting offsets on OffsetOutOfRange errors: <span class=\"name\">earliest\x3c/span> will move to the oldest available message, <span class=\"name\">latest\x3c/span> will move to the most recent. Any other value will raise the exception. Default: <span class=\"name\">latest\x3c/span>.\x3c/li>\n<li><code>enable_auto_commit\x3c/code> (bool): If true the consumer's offset will be periodically committed in the background. Default: <span class=\"name\">True\x3c/span>.\x3c/li>\n<li><code>schema_registry_url\x3c/code> (str): <b>required\x3c/b> for Avro Consumer. Full URL to avro schema endpoint.\x3c/li>\n\x3c/ul>\n<p>Note: Configuration parameters are described in more detail at <a href=\"https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md\">https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md\x3c/a>\x3c/p>","matched":true,"name":"Create Consumer","shortdoc":"Create Kafka Consumer and returns its `group_id` as string.","tags":[]},{"args":["server=127.0.0.1","port=9092","client_id=Robot","group_id=None","schema_registry_url=None","value_schema_str=None","key_schema_str=None","**kwargs"],"doc":"<p>Create Kafka Producer and returns its <span class=\"name\">group_id\x3c/span> as string. If <span class=\"name\">schema_registry_url\x3c/span> is used, Kafka Producer client which does avro schema encoding to messages is created instead.\x3c/p>\n<p>Keyword Arguments:\x3c/p>\n<ul>\n<li><code>server\x3c/code>: (str): IP address / domain, that the consumer should contact to bootstrap initial cluster metadata. Default: <span class=\"name\">127.0.0.1\x3c/span>.\x3c/li>\n<li><code>port\x3c/code> (int): Port number. Default: <span class=\"name\">9092\x3c/span>.\x3c/li>\n<li><code>client_id\x3c/code> (str): a name for this client. This string is passed in each request to servers and can be used to identify specific server-side log entries that correspond to this client. Also submitted to GroupCoordinator for logging with respect to consumer group administration. Default: <span class=\"name\">Robot\x3c/span>.\x3c/li>\n<li><code>group_id\x3c/code> (str or uuid.uuid4() if not set) : name of the consumer group to join for dynamic partition assignment (if enabled), and to use for fetching and committing offsets. If None, unique string is generated  (via uuid.uuid4()) and offset commits are disabled. Default: <span class=\"name\">None\x3c/span>.\x3c/li>\n<li><code>schema_registry_url\x3c/code> (str): <b>required\x3c/b> for Avro Consumer. Full URL to avro schema endpoint.\x3c/li>\n<li><code>value_schema_str\x3c/code> (str): Optional default avro schema for value. Default: <span class=\"name\">None\x3c/span>\x3c/li>\n<li><code>key_schema_str\x3c/code> (str): Optional default avro schema for key. Default: <span class=\"name\">None\x3c/span>\x3c/li>\n\x3c/ul>","matched":true,"name":"Create Producer","shortdoc":"Create Kafka Producer and returns its `group_id` as string. If `schema_registry_url` is used, Kafka Producer client which does avro schema encoding to messages is created instead.","tags":[]},{"args":["data","decode_format","remove_zero_bytes=False"],"doc":"<p>TODO\x3c/p>","matched":true,"name":"Decode Data","shortdoc":"TODO","tags":[]},{"args":["group_id","timeout=1.0"],"doc":"<p>Wait for all messages in the Producer queue to be delivered. This is a convenience method that calls poll() until len() is zero or the optional timeout elapses.\x3c/p>\n<ul>\n<li><span class=\"name\">timeout\x3c/span> (real) : Optional timeout. Default: <span class=\"name\">1.0\x3c/span>.\x3c/li>\n\x3c/ul>","matched":true,"name":"Flush","shortdoc":"Wait for all messages in the Producer queue to be delivered. This is a convenience method that calls poll() until len() is zero or the optional timeout elapses. - `timeout` (real) : Optional timeout. Default: `1.0`.","tags":[]},{"args":["running_thread","decode_format=None","remove_zero_bytes=False"],"doc":"<p>Returns all records gathered from specific thread\x3c/p>\n<ul>\n<li><code>running_thread\x3c/code> (Thread object) - thread which was executed by <a href=\"#Start%20Consumer%20Threaded\" class=\"name\">Start Consumer Threaded\x3c/a>\x3c/li>\n<li><code>decode_data\x3c/code> (str) - If you need to decode data to specific format (See <a href=\"https://docs.python.org/3/library/codecs.html#standard-encodings\">https://docs.python.org/3/library/codecs.html#standard-encodings\x3c/a>). Default: None.\x3c/li>\n<li><code>remove_zero_bytes\x3c/code> (bool) - When you are working with byte streams you can end up with a lot of '\\x00' bytes you want to remove. Default: False.\x3c/li>\n\x3c/ul>","matched":true,"name":"Get Messages From Thread","shortdoc":"Returns all records gathered from specific thread - ``running_thread`` (Thread object) - thread which was executed by `Start Consumer Threaded` - ``decode_data`` (str) - If you need to decode data to specific format     (See https://docs.python.org/3/library/codecs.html#standard-encodings). Default: None. - ``remove_zero_bytes`` (bool) - When you are working with byte streams     you can end up with a lot of '\\x00' bytes you want to remove. Default: False.","tags":[]},{"args":["group_id"],"doc":"","matched":true,"name":"List Topics","shortdoc":"","tags":[]},{"args":["group_id","timeout=1","max_records=1","poll_attempts=10"],"doc":"<p>Fetch and return messages from assigned topics / partitions as list.\x3c/p>\n<ul>\n<li><code>max_records\x3c/code> (int): maximum number of messages to get from poll. Default: 1.\x3c/li>\n<li><code>timeout\x3c/code> (int): Seconds spent waiting in poll if data is not available in the buffer.\x3c/li>\n\x3c/ul>\n<p>If 0, returns immediately with any records that are available currently in the buffer, else returns empty. Must not be negative. Default: <span class=\"name\">1\x3c/span>\x3c/p>\n<ul>\n<li><code>poll_attempts\x3c/code> (int): Attempts to consume messages and endless looping prevention.\x3c/li>\n\x3c/ul>\n<p>Sometimes the first messages are None or the topic could be empty. Default: <span class=\"name\">10\x3c/span>.\x3c/p>","matched":true,"name":"Poll","shortdoc":"Fetch and return messages from assigned topics / partitions as list. - ``max_records`` (int): maximum number of messages to get from poll. Default: 1. - ``timeout`` (int): Seconds spent waiting in poll if data is not available in the buffer. If 0, returns immediately with any records that are available currently in the buffer, else returns empty. Must not be negative. Default: `1` -  ``poll_attempts`` (int): Attempts to consume messages and endless looping prevention. Sometimes the first messages are None or the topic could be empty. Default: `10`.","tags":[]},{"args":["group_id","topic","value","key=None","value_encoding=utf-8","callback_func=None"],"doc":"<p>Produce message to topic asynchronously to Kafka by encoding with specified or default avro schema.\x3c/p>\n<ul>\n<li><span class=\"name\">`topic\x3c/span> (str) : name of the topic where to produce message.\x3c/li>\n<li><span class=\"name\">`data\x3c/span> (str) : Message which produce to topic.\x3c/li>\n<li><span class=\"name\">`data_encoding\x3c/span> (str) : encode to specific format. Default: <span class=\"name\">utf-8\x3c/span>.\x3c/li>\n<li><span class=\"name\">`callback_func\x3c/span> (func) : the function that will be called from poll()\x3c/li>\n<li><code>key\x3c/code> (object) : An object to serialize. Default: <span class=\"name\">None\x3c/span>. when the message has been successfully delivered or permanently fails delivery..\x3c/li>\n\x3c/ul>","matched":true,"name":"Produce","shortdoc":"Produce message to topic asynchronously to Kafka by encoding with specified or default avro schema.","tags":[]},{"args":["topics","group_id=None","server=127.0.0.1","port=9092","**kwargs"],"doc":"","matched":true,"name":"Start Consumer Threaded","shortdoc":"","tags":[]},{"args":["running_thread"],"doc":"","matched":true,"name":"Stop Thread","shortdoc":"","tags":[]},{"args":["group_id","topics"],"doc":"<p>Subscribe to a list of topics, or a topic regex pattern.\x3c/p>\n<ul>\n<li><code>topics\x3c/code> (list): List of topics for subscription.\x3c/li>\n\x3c/ul>","matched":true,"name":"Subscribe Topic","shortdoc":"Subscribe to a list of topics, or a topic regex pattern.","tags":[]},{"args":["group_id"],"doc":"<p>Unsubscribe of topics.\x3c/p>","matched":true,"name":"Unsubscribe","shortdoc":"Unsubscribe of topics.","tags":[]}],"name":"ConfluentKafkaLibrary","named_args":true,"scope":"global","version":"0.3.0"};
+libdoc = {"all_tags":[],"contains_tags":false,"doc":"<p>ConfluentKafkaLibrary is a Robot Framework library which wraps up <a href=\"https://github.com/confluentinc/confluent-kafka-python\">confluent-kafka-python\x3c/a>. Keywords are inspired by <a href=\"https://github.com/s4int/robotframework-KafkaLibrary\">robotframework-KafkaLibrary\x3c/a> which is based on <a href=\"https://github.com/Parsely/pykafka\">pykafka project\x3c/a> and support more functionality like running more clients based on <span class=\"name\">group_id\x3c/span> or running them in threaded mode during the tests, decoding of gathered data etc. (<span class=\"name\">See\x3c/span> <a href=\"#Examples\" class=\"name\">Examples\x3c/a>).\x3c/p>\n<p>This document explains how to use keywords provided by ConfluentKafkaLibrary. For information about installation, support, and more, please visit the <a href=\"https://github.com/robooo/robotframework-ConfluentKafkaLibrary\">project github page\x3c/a>. For more information about Robot Framework, see <a href=\"http://robotframework.org\">http://robotframework.org\x3c/a>.\x3c/p>\n<h3 id=\"Examples\">Examples\x3c/h3>\n<p><b>Basic Consumer with predefined group_id\x3c/b>\x3c/p>\n<table border=\"1\">\n<tr>\n<td>${group_id}=\x3c/td>\n<td><a href=\"#Create%20Consumer\" class=\"name\">Create Consumer\x3c/a>\x3c/td>\n<td>group_id=mygroup\x3c/td>\n<td># if group_id is not defined uuid4() is gemerated\x3c/td>\n\x3c/tr>\n<tr>\n<td><a href=\"#Subscribe%20Topic\" class=\"name\">Subscribe Topic\x3c/a>\x3c/td>\n<td>group_id=${group_id}\x3c/td>\n<td>topics=test_topic\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n<tr>\n<td>${result}=\x3c/td>\n<td><a href=\"#Poll\" class=\"name\">Poll\x3c/a>\x3c/td>\n<td>group_id=${group_id}\x3c/td>\n<td>max_records=5\x3c/td>\n\x3c/tr>\n<tr>\n<td><span class=\"name\">Log\x3c/span>\x3c/td>\n<td>${result}\x3c/td>\n<td>\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n<tr>\n<td><a href=\"#Unsubscribe\" class=\"name\">Unsubscribe\x3c/a>\x3c/td>\n<td>${group_id}\x3c/td>\n<td>\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n<tr>\n<td><a href=\"#Close%20Consumer\" class=\"name\">Close Consumer\x3c/a>\x3c/td>\n<td>${group_id}\x3c/td>\n<td>\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n\x3c/table>\n<p><b>More Consumers\x3c/b>\x3c/p>\n<table border=\"1\">\n<tr>\n<td>${group_id_1}=\x3c/td>\n<td><a href=\"#Create%20Consumer\" class=\"name\">Create Consumer\x3c/a>\x3c/td>\n<td>\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n<tr>\n<td><a href=\"#Subscribe%20Topic\" class=\"name\">Subscribe Topic\x3c/a>\x3c/td>\n<td>group_id=${group_id_1}\x3c/td>\n<td>topics=topic1\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n<tr>\n<td>${group_id_2}=\x3c/td>\n<td><a href=\"#Create%20Consumer\" class=\"name\">Create Consumer\x3c/a>\x3c/td>\n<td>\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n<tr>\n<td><a href=\"#Subscribe%20Topic\" class=\"name\">Subscribe Topic\x3c/a>\x3c/td>\n<td>group_id=${group_id_2}\x3c/td>\n<td>topics=topic2\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n<tr>\n<td>${result_1}=\x3c/td>\n<td><a href=\"#Poll\" class=\"name\">Poll\x3c/a>\x3c/td>\n<td>group_id=${group_id_1}\x3c/td>\n<td>max_records=5\x3c/td>\n\x3c/tr>\n<tr>\n<td>${result_2}=\x3c/td>\n<td><a href=\"#Poll\" class=\"name\">Poll\x3c/a>\x3c/td>\n<td>group_id=${group_id_2}\x3c/td>\n<td>max_records=2\x3c/td>\n\x3c/tr>\n<tr>\n<td><a href=\"#Unsubscribe\" class=\"name\">Unsubscribe\x3c/a>\x3c/td>\n<td>${group_id_1}\x3c/td>\n<td>\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n<tr>\n<td><a href=\"#Unsubscribe\" class=\"name\">Unsubscribe\x3c/a>\x3c/td>\n<td>${group_id_2}\x3c/td>\n<td>\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n<tr>\n<td><a href=\"#Close%20Consumer\" class=\"name\">Close Consumer\x3c/a>\x3c/td>\n<td>${group_id_1}\x3c/td>\n<td>\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n<tr>\n<td><a href=\"#Close%20Consumer\" class=\"name\">Close Consumer\x3c/a>\x3c/td>\n<td>${group_id_2}\x3c/td>\n<td>\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n\x3c/table>\n<p><b>Run Avro Consumer over HTTPS and threaded:\x3c/b>\x3c/p>\n<table border=\"1\">\n<tr>\n<td>${thread}=\x3c/td>\n<td><a href=\"#Start%20Consumer%20Threaded\" class=\"name\">Start Consumer Threaded\x3c/a>\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n<tr>\n<td>\x3c/td>\n<td>...  topics=test_topic\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n<tr>\n<td>\x3c/td>\n<td>...  schema_registry_url=https://localhost:8081\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n<tr>\n<td>\x3c/td>\n<td>...  auto_offset_reset=earliest\x3c/td>\n<td># We want to get all messages\x3c/td>\n\x3c/tr>\n<tr>\n<td>\x3c/td>\n<td>...  security.protocol=ssl\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n<tr>\n<td>\x3c/td>\n<td>...  schema.registry.ssl.ca.location=/home/user/cert/caproxy.pem\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n<tr>\n<td>\x3c/td>\n<td>...  ssl.ca.location=/home/user/cert/caproxy.pem\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n<tr>\n<td>\x3c/td>\n<td>...  ssl.certificate.location=/home/user/cert/kafka-client-cert.pem\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n<tr>\n<td>\x3c/td>\n<td>...  ssl.key.location=/home/user/cert/kafka-client-key.pem\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n<tr>\n<td><span class=\"name\">Log\x3c/span>\x3c/td>\n<td><span class=\"name\">Execute commands which should push some data to topic\x3c/span>\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n<tr>\n<td>${messages}=\x3c/td>\n<td><a href=\"#Get%20Messages%20From%20Thread\" class=\"name\">Get Messages From Thread\x3c/a>\x3c/td>\n<td>${thread}\x3c/td>\n\x3c/tr>\n<tr>\n<td>Stop Thread\x3c/td>\n<td>${thread}\x3c/td>\n<td>\x3c/td>\n\x3c/tr>\n\x3c/table>","generated":"2020-04-11 13:52:11","inits":[],"keywords":[{"args":["group_id","topic_partitions"],"doc":"<p>Assign a list of TopicPartitions.\x3c/p>\n<ul>\n<li><code>topic_partitions\x3c/code> (<span class=\"name\">TopicPartition\x3c/span> or list of <span class=\"name\">TopicPartition\x3c/span>): Assignment for this instance.\x3c/li>\n\x3c/ul>","matched":true,"name":"Assign To Topic Partition","shortdoc":"Assign a list of TopicPartitions.","tags":[]},{"args":["running_thread"],"doc":"<p>Remove all records gathered from specific thread\x3c/p>\n<ul>\n<li><code>running_thread\x3c/code> (Thread object) - thread which was executed with <a href=\"#Start%20Consumer%20Threaded\" class=\"name\">Start Consumer Threaded\x3c/a> keyword\x3c/li>\n\x3c/ul>","matched":true,"name":"Clear Messages From Thread","shortdoc":"Remove all records gathered from specific thread - ``running_thread`` (Thread object) - thread which was executed with     `Start Consumer Threaded` keyword","tags":[]},{"args":["group_id"],"doc":"<p>Close down and terminate the Kafka Consumer.\x3c/p>","matched":true,"name":"Close Consumer","shortdoc":"Close down and terminate the Kafka Consumer.","tags":[]},{"args":["group_id=None","server=127.0.0.1","port=9092","enable_auto_commit=True","auto_offset_reset=latest","schema_registry_url=None","**kwargs"],"doc":"<p>Create Kafka Consumer and returns its <span class=\"name\">group_id\x3c/span> as string.\x3c/p>\n<p>Keyword Arguments:\x3c/p>\n<ul>\n<li><code>server\x3c/code>: (str): IP address / domain, that the consumer should contact to bootstrap initial cluster metadata. Default: <span class=\"name\">127.0.0.1\x3c/span>.\x3c/li>\n<li><code>port\x3c/code> (int): Port number. Default: <span class=\"name\">9092\x3c/span>.\x3c/li>\n<li><code>group_id\x3c/code> (str or uuid.uuid4() if not set) : name of the consumer group to join for dynamic partition assignment (if enabled), and to use for fetching and committing offsets. If None, unique string is generated  (via uuid.uuid4()) and offset commits are disabled. Default: <span class=\"name\">None\x3c/span>.\x3c/li>\n<li><code>auto_offset_reset\x3c/code> (str): A policy for resetting offsets on OffsetOutOfRange errors: <span class=\"name\">earliest\x3c/span> will move to the oldest available message, <span class=\"name\">latest\x3c/span> will move to the most recent. Any other value will raise the exception. Default: <span class=\"name\">latest\x3c/span>.\x3c/li>\n<li><code>enable_auto_commit\x3c/code> (bool): If true the consumer's offset will be periodically committed in the background. Default: <span class=\"name\">True\x3c/span>.\x3c/li>\n<li><code>schema_registry_url\x3c/code> (str): <b>required\x3c/b> for Avro Consumer. Full URL to avro schema endpoint.\x3c/li>\n\x3c/ul>\n<p>Note: Configuration parameters are described in more detail at <a href=\"https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md\">https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md\x3c/a>\x3c/p>","matched":true,"name":"Create Consumer","shortdoc":"Create Kafka Consumer and returns its `group_id` as string.","tags":[]},{"args":["server=127.0.0.1","port=9092","client_id=Robot","group_id=None","schema_registry_url=None","value_schema=None","key_schema=None","**kwargs"],"doc":"<p>Create Kafka Producer and returns its <span class=\"name\">group_id\x3c/span> as string. If <span class=\"name\">schema_registry_url\x3c/span> is used, Kafka Producer client which does avro schema encoding to messages is created instead.\x3c/p>\n<p>Keyword Arguments:\x3c/p>\n<ul>\n<li><code>server\x3c/code>: (str): IP address / domain, that the consumer should contact to bootstrap initial cluster metadata. Default: <span class=\"name\">127.0.0.1\x3c/span>.\x3c/li>\n<li><code>port\x3c/code> (int): Port number. Default: <span class=\"name\">9092\x3c/span>.\x3c/li>\n<li><code>client_id\x3c/code> (str): a name for this client. This string is passed in each request to servers and can be used to identify specific server-side log entries that correspond to this client. Also submitted to GroupCoordinator for logging with respect to consumer group administration. Default: <span class=\"name\">Robot\x3c/span>.\x3c/li>\n<li><code>group_id\x3c/code> (str or uuid.uuid4() if not set) : name of the consumer group to join for dynamic partition assignment (if enabled), and to use for fetching and committing offsets. If None, unique string is generated  (via uuid.uuid4()) and offset commits are disabled. Default: <span class=\"name\">None\x3c/span>.\x3c/li>\n<li><code>schema_registry_url\x3c/code> (str): <b>required\x3c/b> for Avro Consumer. Full URL to avro schema endpoint.\x3c/li>\n<li><code>value_schema\x3c/code> (str): Optional default avro schema for value or path to file with schema. Default: <span class=\"name\">None\x3c/span>\x3c/li>\n<li><code>key_schema\x3c/code> (str): Optional default avro schema for key. Default: <span class=\"name\">None\x3c/span>\x3c/li>\n\x3c/ul>","matched":true,"name":"Create Producer","shortdoc":"Create Kafka Producer and returns its `group_id` as string. If `schema_registry_url` is used, Kafka Producer client which does avro schema encoding to messages is created instead.","tags":[]},{"args":["topic_name","partition=None","offset=None"],"doc":"<p>Returns TopicPartiton object based on <a href=\"https://docs.confluent.io/current/clients/confluent-kafka-python/#topicpartition\">https://docs.confluent.io/current/clients/confluent-kafka-python/#topicpartition\x3c/a>\x3c/p>\n<ul>\n<li><code>topic_name\x3c/code> (str): Topic name.\x3c/li>\n<li><code>partition\x3c/code> (int): Partition id.\x3c/li>\n<li><code>offset\x3c/code> (int): Initial partition offset.\x3c/li>\n\x3c/ul>","matched":true,"name":"Create Topic Partition","shortdoc":"Returns TopicPartiton object based on https://docs.confluent.io/current/clients/confluent-kafka-python/#topicpartition","tags":[]},{"args":["group_id","timeout=0.1"],"doc":"<p>Wait for all messages in the Producer queue to be delivered. Returns the number of messages still in queue. This is a convenience method that calls poll() until len() is zero or the optional timeout elapses.\x3c/p>\n<ul>\n<li><span class=\"name\">timeout\x3c/span> (real) : Optional timeout. Default: <span class=\"name\">0.1\x3c/span>.\x3c/li>\n\x3c/ul>","matched":true,"name":"Flush","shortdoc":"Wait for all messages in the Producer queue to be delivered. Returns the number of messages still in queue. This is a convenience method that calls poll() until len() is zero or the optional timeout elapses. - `timeout` (real) : Optional timeout. Default: `0.1`.","tags":[]},{"args":["group_id"],"doc":"","matched":true,"name":"Get Assignment","shortdoc":"","tags":[]},{"args":["running_thread","decode_format=None"],"doc":"<p>Returns all records gathered from specific thread\x3c/p>\n<ul>\n<li><code>running_thread\x3c/code> (Thread object) - thread which was executed with <a href=\"#Start%20Consumer%20Threaded\" class=\"name\">Start Consumer Threaded\x3c/a> keyword\x3c/li>\n<li><code>decode_format\x3c/code> (str) - If you need to decode data to specific format (See <a href=\"https://docs.python.org/3/library/codecs.html#standard-encodings\">https://docs.python.org/3/library/codecs.html#standard-encodings\x3c/a>). Default: None.\x3c/li>\n\x3c/ul>","matched":true,"name":"Get Messages From Thread","shortdoc":"Returns all records gathered from specific thread - ``running_thread`` (Thread object) - thread which was executed with     `Start Consumer Threaded` keyword - ``decode_format`` (str) - If you need to decode data to specific format     (See https://docs.python.org/3/library/codecs.html#standard-encodings). Default: None.","tags":[]},{"args":["group_id","topic_partitions"],"doc":"<p>Retrieve current positions (offsets) for the list of partitions.\x3c/p>\n<ul>\n<li><code>topic_partitions\x3c/code> (<span class=\"name\">TopicPartition\x3c/span> or list of <span class=\"name\">TopicPartition\x3c/span>): Assignment for this instance.\x3c/li>\n\x3c/ul>","matched":true,"name":"Get Position","shortdoc":"Retrieve current positions (offsets) for the list of partitions.","tags":[]},{"args":["topic"],"doc":"<p>Returns dictionary of all TopicPartitons in topic (topic.partitions).\x3c/p>","matched":true,"name":"Get Topic Partitions","shortdoc":"Returns dictionary of all TopicPartitons in topic (topic.partitions).","tags":[]},{"args":["group_id","topic_partition","**kwargs"],"doc":"<p>Retrieve low and high offsets for partition.\x3c/p>","matched":true,"name":"Get Watermark Offsets","shortdoc":"Retrieve low and high offsets for partition.","tags":[]},{"args":["group_id","topic=None"],"doc":"<p>Request Metadata from cluster. Could be executed with consumer or producer group_id too.\x3c/p>\n<ul>\n<li><code>topic\x3c/code> (str):  If specified, only request info about this topic, else returnfor all topics in cluster.\x3c/li>\n\x3c/ul>\n<p>Default: <span class=\"name\">None\x3c/span>.\x3c/p>\n<ul>\n<li><code>group_id\x3c/code> (str): <b>required\x3c/b> id of the created consumer or producer.\x3c/li>\n\x3c/ul>","matched":true,"name":"List Topics","shortdoc":"Request Metadata from cluster. Could be executed with consumer or producer group_id too. - ``topic`` (str):  If specified, only request info about this topic, else returnfor all topics in cluster. Default: `None`. - ``group_id`` (str): *required* id of the created consumer or producer.","tags":[]},{"args":["data"],"doc":"","matched":true,"name":"Load Schema","shortdoc":"","tags":[]},{"args":["group_id","timeout=1","max_records=1","poll_attempts=10","only_value=True","decode_format=None"],"doc":"<p>Fetch and return messages from assigned topics / partitions as list.\x3c/p>\n<ul>\n<li><code>timeout\x3c/code> (int): Seconds spent waiting in poll if data is not available in the buffer.\x3c/li>\n\x3c/ul>\n<ul>\n<li><code>max_records\x3c/code> (int): maximum number of messages to get from poll. Default: 1.\x3c/li>\n\x3c/ul>\n<p>If 0, returns immediately with any records that are available currently in the buffer, else returns empty. Must not be negative. Default: <span class=\"name\">1\x3c/span>\x3c/p>\n<ul>\n<li><code>poll_attempts\x3c/code> (int): Attempts to consume messages and endless looping prevention.\x3c/li>\n\x3c/ul>\n<p>Sometimes the first messages are None or the topic could be empty. Default: <span class=\"name\">10\x3c/span>.\x3c/p>\n<ul>\n<li><code>only_value\x3c/code> (bool): Return only message.value(). Default: <span class=\"name\">True\x3c/span>.\x3c/li>\n<li><code>decode_format\x3c/code> (str) - If you need to decode data to specific format (See <a href=\"https://docs.python.org/3/library/codecs.html#standard-encodings\">https://docs.python.org/3/library/codecs.html#standard-encodings\x3c/a>). Default: None.\x3c/li>\n\x3c/ul>","matched":true,"name":"Poll","shortdoc":"Fetch and return messages from assigned topics / partitions as list. - ``timeout`` (int): Seconds spent waiting in poll if data is not available in the buffer.","tags":[]},{"args":["group_id","topic","value=None","key=None","**kwargs"],"doc":"<p>Produce message to topic asynchronously to Kafka by encoding with specified or default avro schema.\x3c/p>\n<p><a href=\"https://docs.confluent.io/current/clients/confluent-kafka-python/#confluent_kafka.Producer.produce\">https://docs.confluent.io/current/clients/confluent-kafka-python/#confluent_kafka.Producer.produce\x3c/a>\x3c/p>\n<ul>\n<li><code>topic\x3c/code> (str) : name of the topic where to produce message.\x3c/li>\n<li><code>value\x3c/code> (str|bytes): Message payload.\x3c/li>\n<li><code>key\x3c/code> (str|bytes): Message key. Default: <span class=\"name\">None\x3c/span>.\x3c/li>\n<li><code>partition\x3c/code> (int): Partition to produce to, else uses the configured built-in partitioner.\x3c/li>\n\x3c/ul>","matched":true,"name":"Produce","shortdoc":"Produce message to topic asynchronously to Kafka by encoding with specified or default avro schema.","tags":[]},{"args":["group_id","topic_partition"],"doc":"<p><a href=\"https://docs.confluent.io/current/clients/confluent-kafka-python/#confluent_kafka.Consumer.seek\">https://docs.confluent.io/current/clients/confluent-kafka-python/#confluent_kafka.Consumer.seek\x3c/a>\x3c/p>","matched":true,"name":"Seek","shortdoc":"https://docs.confluent.io/current/clients/confluent-kafka-python/#confluent_kafka.Consumer.seek","tags":[]},{"args":["topics","group_id=None","server=127.0.0.1","port=9092","only_value=True","**kwargs"],"doc":"<p>Run consumer in daemon thread and store data from topics. To read and work with this collected data use keyword <a href=\"#Get%20Messages%20From%20Thread\" class=\"name\">Get Messages From Thread\x3c/a>. Could be used at the Test setup or in each test. This is useful when you are reading always the same topics and you don't want to create consumer in each test to poll data. You can create as many consumers in the Test setup as you want and then in test just read data with <a href=\"#Get%20Messages%20From%20Thread\" class=\"name\">Get Messages From Thread\x3c/a> keyword.\x3c/p>\n<ul>\n<li><code>topics\x3c/code> (list): List of topics for subscription.\x3c/li>\n<li><code>group_id\x3c/code> (str or uuid.uuid4() if not set) : name of the consumer group to join for dynamic partition assignment (if enabled), and to use for fetching and committing offsets. If None, unique string is generated  (via uuid.uuid4()) and offset commits are disabled. Default: <span class=\"name\">None\x3c/span>.\x3c/li>\n\x3c/ul>","matched":true,"name":"Start Consumer Threaded","shortdoc":"Run consumer in daemon thread and store data from topics. To read and work with this    collected data use keyword `Get Messages From Thread`.    Could be used at the Test setup or in each test.    This is useful when you are reading always the same topics and you don't want to create    consumer in each test to poll data. You can create as many consumers in the Test setup    as you want and then in test just read data with `Get Messages From Thread` keyword. - ``topics`` (list): List of topics for subscription. - ``group_id`` (str or uuid.uuid4() if not set) : name of the consumer group to join for     dynamic partition assignment (if enabled), and to use for fetching and     committing offsets. If None, unique string is generated  (via uuid.uuid4())     and offset commits are disabled. Default: `None`.","tags":[]},{"args":["group_id","topics","**kwargs"],"doc":"<p>Subscribe to a list of topics, or a topic regex pattern. <a href=\"https://docs.confluent.io/current/clients/confluent-kafka-python/#confluent_kafka.Consumer.subscribe\">https://docs.confluent.io/current/clients/confluent-kafka-python/#confluent_kafka.Consumer.subscribe\x3c/a>\x3c/p>\n<ul>\n<li><code>topics\x3c/code> (list): List of topics for subscription.\x3c/li>\n\x3c/ul>","matched":true,"name":"Subscribe Topic","shortdoc":"Subscribe to a list of topics, or a topic regex pattern.    https://docs.confluent.io/current/clients/confluent-kafka-python/#confluent_kafka.Consumer.subscribe","tags":[]},{"args":["group_id"],"doc":"","matched":true,"name":"Unassign","shortdoc":"","tags":[]},{"args":["group_id"],"doc":"<p>Unsubscribe of topics.\x3c/p>","matched":true,"name":"Unsubscribe","shortdoc":"Unsubscribe of topics.","tags":[]}],"name":"ConfluentKafkaLibrary","named_args":true,"scope":"global","version":"0.3.0"};
 </script>
 <title></title>
 </head>
diff --git a/examples/docker-compose.yml b/examples/docker-compose.yml
new file mode 100644
index 0000000..830ef2c
--- /dev/null
+++ b/examples/docker-compose.yml
@@ -0,0 +1,56 @@
+---
+    version: '2'
+    services:
+      zookeeper:
+        image: confluentinc/cp-zookeeper:5.3.1
+        hostname: zookeeper
+        container_name: zookeeper
+        ports:
+          - "2181:2181"
+        environment:
+          ZOOKEEPER_CLIENT_PORT: 2181
+    
+      broker:
+        image: confluentinc/cp-enterprise-kafka:5.3.1
+        hostname: broker
+        container_name: broker
+        depends_on:
+          - zookeeper
+        ports:
+          - "9092:9092"
+        environment:
+          KAFKA_BROKER_ID: 1
+          KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
+          KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
+          KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092
+          KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+          KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
+    
+      schema-registry:
+        image: confluentinc/cp-schema-registry:5.3.1
+        hostname: schema-registry
+        container_name: schema-registry
+        depends_on:
+          - zookeeper
+          - broker
+        ports:
+          - "8081:8081"
+        environment:
+          SCHEMA_REGISTRY_HOST_NAME: schema-registry
+          SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: 'zookeeper:2181'
+
+      rest-proxy:
+        image: confluentinc/cp-kafka-rest:5.3.1
+        depends_on:
+            - zookeeper
+            - broker
+            - schema-registry
+        ports:
+            - 8082:8082
+        hostname: rest-proxy
+        container_name: rest-proxy
+        environment:
+            KAFKA_REST_HOST_NAME: rest-proxy
+            KAFKA_REST_BOOTSTRAP_SERVERS: 'broker:29092'
+            KAFKA_REST_LISTENERS: "http://0.0.0.0:8082"
+            KAFKA_REST_SCHEMA_REGISTRY_URL: 'http://schema-registry:8081'
diff --git a/examples/schema/producer/KeySchema.avsc b/examples/schema/producer/KeySchema.avsc
new file mode 100644
index 0000000..1436583
--- /dev/null
+++ b/examples/schema/producer/KeySchema.avsc
@@ -0,0 +1,11 @@
+{
+  "namespace": "example.avro",
+  "type": "record",
+  "name": "User",
+  "fields": [
+    {
+      "name": "name",
+      "type": "string"
+    }
+  ]
+}
\ No newline at end of file
diff --git a/examples/schema/producer/ValueSchema.avsc b/examples/schema/producer/ValueSchema.avsc
new file mode 100644
index 0000000..9a33eb5
--- /dev/null
+++ b/examples/schema/producer/ValueSchema.avsc
@@ -0,0 +1,18 @@
+{
+  "namespace": "example.avro",
+  "type": "record",
+  "name": "User",
+  "fields": [
+    {
+      "name": "name",
+      "type": "string"
+    },
+    {
+      "name": "number",
+      "type": [
+        "int",
+        "null"
+      ]
+    }
+  ]
+}
\ No newline at end of file
diff --git a/examples/test.robot b/examples/test.robot
index 5136c0d..c6907cd 100644
--- a/examples/test.robot
+++ b/examples/test.robot
@@ -6,32 +6,163 @@ Suite Setup  Starting Test
 
 
 *** Test Cases ***
+Verify Topics
+    ${group_id}=  Create Consumer  auto_offset_reset=earliest
+    ${topics}=  List Topics  ${group_id}
+    Dictionary Should Contain Key  ${topics}  ${TEST_TOPIC}
+    [Teardown]  Close Consumer  ${group_id}
+
 Basic Consumer
     ${group_id}=  Create Consumer  auto_offset_reset=earliest
-    Subscribe Topic  group_id=${group_id}  topics=test
-    ${messages}=  Poll  group_id=${group_id}  max_records=3
-    ${messages}=  Decode Data  ${messages}  decode_format=utf8
+    Subscribe Topic  group_id=${group_id}  topics=${TEST_TOPIC}
+    ${messages}=  Poll  group_id=${group_id}  max_records=3  decode_format=utf8
     ${data}=  Create List  Hello  World  {'test': 1}
     Lists Should Be Equal  ${messages}  ${data}
-    Unsubscribe  ${group_id}
-    Close Consumer  ${group_id}
+    [Teardown]  Basic Teardown  ${group_id}
 
-Verify Threaded Consumer
-    ${thread_messages}=  Get Messages From Thread  ${MAIN_THREAD}
+Produce Without Value
+    ${topic_name}=  Set Variable  topicwithoutvaluee
+    Produce  group_id=${PRODUCER_ID}  topic=${topic_name}
+    Wait Until Keyword Succeeds  10x  0.5s  All Messages Are Delivered  ${PRODUCER_ID}
     ${group_id}=  Create Consumer  auto_offset_reset=earliest
-    Subscribe Topic  group_id=${group_id}  topics=test
-    ${messages}=  Poll  group_id=${group_id}  max_records=3
-    List Should Contain Sub List  ${thread_messages}  ${messages}
-    Unsubscribe  ${group_id}
-    Close Consumer  ${group_id}
+    Subscribe Topic  group_id=${group_id}  topics=${topic_name}
+    ${messages}=  Poll  group_id=${group_id}  max_records=1
+    Should Be Equal As Strings  ${messages}  [None]
+
+Verify Position
+    ${group_id}=  Create Consumer
+    ${tp}=  Create Topic Partition  ${TEST_TOPIC}  ${P_ID}  ${OFFSET_END}
+    Assign To Topic Partition  ${group_id}  ${tp}
+    Sleep  5sec  # Need to wait for an assignment
+    ${position}=  Get Position  group_id=${group_id}  topic_partitions=${tp}
+    ${position_before}=  Set Variable  ${position[0].offset}
+
+    Produce  group_id=${PRODUCER_ID}  topic=${TEST_TOPIC}  value=Dummy  partition=${P_ID}
+    Wait Until Keyword Succeeds  10x  0.5s  All Messages Are Delivered  ${PRODUCER_ID}
+    ${position}=  Get Position  group_id=${group_id}  topic_partitions=${tp}
+    ${position_after_produce}=  Set Variable  ${position[0].offset}
+    Should Be Equal As Integers  ${position_before}  ${position_after_produce}
+
+    ${messages}=  Poll  group_id=${group_id}  max_records=1  decode_format=utf8
+    ${position}=  Get Position  group_id=${group_id}  topic_partitions=${tp}
+    ${position_after_poll_1}=  Set Variable  ${position[0].offset}
+    Should Not Be Equal As Integers  ${position_after_poll_1}  ${position_after_produce}
+
+    Produce  group_id=${PRODUCER_ID}  topic=${TEST_TOPIC}  value=Dummy  partition=${P_ID}
+    Wait Until Keyword Succeeds  10x  0.5s  All Messages Are Delivered  ${PRODUCER_ID}
+    ${messages}=  Poll  group_id=${group_id}  max_records=1  decode_format=utf8
+    ${position}=  Get Position  group_id=${group_id}  topic_partitions=${tp}
+    ${position_after_poll_2}=  Set Variable  ${position[0].offset}
+    Should Be Equal As Integers  ${position_after_poll_1 + 1}  ${position_after_poll_2}
+    [Teardown]  Basic Teardown  ${group_id}
+
+Consumer With Assignment To Last Message After Get Of Watermark Offsets
+    ${group_id}=  Create Consumer
+    ${tp}=  Create Topic Partition  ${TEST_TOPIC}  ${P_ID}
+    ${offset}=  Get Watermark Offsets  ${group_id}  ${tp}
+    ${tp}=  Create Topic Partition  ${TEST_TOPIC}  ${P_ID}  ${offset[1]}
+    Assign To Topic Partition  ${group_id}  ${tp}
+    Prepare Data
+    ${messages}=  Poll  group_id=${group_id}  max_records=6  decode_format=utf8
+    Lists Should Be Equal  ${TEST_DATA}  ${messages}
+    [Teardown]  Basic Teardown  ${group_id}
+
+Consumer With Assignment To OFFSET_END
+    ${group_id}=  Create Consumer
+    ${tp}=  Create Topic Partition  ${TEST_TOPIC}  ${P_ID}  ${OFFSET_END}
+    Assign To Topic Partition  ${group_id}  ${tp}
+    # Need to wait for an async assignment, be aware the Is Assigned could return True but
+    # that doesn't mean assignment is completed
+    Sleep  5sec
+    Prepare Data
+    ${messages}=  Poll  group_id=${group_id}  poll_attempts=30  max_records=6  timeout=5  decode_format=utf8
+    Lists Should Be Equal  ${TEST_DATA}  ${messages}
+    [Teardown]  Unassign Teardown  ${group_id}
+
+Verify Test And Threaded Consumer
+    [Setup]  Clear Messages From Thread  ${MAIN_THREAD}
+    ${group_id}=  Create Consumer
+    Subscribe Topic  group_id=${group_id}  topics=${TEST_TOPIC}
+    ${messages}=  Poll  group_id=${group_id}
+    Prepare Data
+    ${thread_messages}=  Get Messages From Thread  ${MAIN_THREAD}  decode_format=utf-8
+    ${messages}=  Poll  group_id=${group_id}  max_records=6  decode_format=utf8
+    Lists Should Be Equal  ${thread_messages}  ${messages}
+    [Teardown]  Run Keywords  Basic Teardown  ${group_id}  AND
+                ...  Clear Messages From Thread  ${MAIN_THREAD}
+
+Verify Clean Of Threaded Consumer Messages
+    [Setup]  Prepare Data
+    ${thread_messages1}=  Get Messages From Thread  ${MAIN_THREAD}  decode_format=utf-8
+    Clear Messages From Thread  ${MAIN_THREAD}
+    ${thread_messages2}=  Get Messages From Thread  ${MAIN_THREAD}
+    Lists Should Be Equal  ${TEST_DATA}  ${thread_messages1}
+    Should Be Empty  ${thread_messages2}
+    [Teardown]  Clear Messages From Thread  ${MAIN_THREAD}
+
+Remove And Publish New Messages From Threaded Consumer
+    [Setup]  Prepare Data
+    ${thread_messages1}=  Get Messages From Thread  ${MAIN_THREAD}  decode_format=utf-8
+    Clear Messages From Thread  ${MAIN_THREAD}
+    Produce  group_id=${PRODUCER_ID}  topic=${TEST_TOPIC}  value=After  partition=${P_ID}
+    Produce  group_id=${PRODUCER_ID}  topic=${TEST_TOPIC}  value=Clear  partition=${P_ID}
+    Wait Until Keyword Succeeds  10x  0.5s  All Messages Are Delivered  ${PRODUCER_ID}
+    Sleep  1sec  # if next command is polling messages in thread we need to wait a second
+
+    ${thread_messages2}=  Get Messages From Thread  ${MAIN_THREAD}  decode_format=utf-8
+    ${data}=  Create List  After  Clear
+    Should Be Equal  ${data}  ${thread_messages2}
+
+    Produce  group_id=${PRODUCER_ID}  topic=${TEST_TOPIC}  value=LAST  partition=${P_ID}
+    Wait Until Keyword Succeeds  10x  0.5s  All Messages Are Delivered  ${PRODUCER_ID}
+    Sleep  1sec
+    Append To List  ${data}  LAST
+    ${thread_messages2}=  Get Messages From Thread  ${MAIN_THREAD}  decode_format=utf-8
+    Should Be Equal  ${TEST_DATA}  ${thread_messages1}
+    Should Be Equal  ${data}  ${thread_messages2}
+    [Teardown]  Clear Messages From Thread  ${MAIN_THREAD}
+
 
 *** Keywords ***
 Starting Test
-    ${thread}=  Start Consumer Threaded  topics=test  auto_offset_reset=earliest
-    Set Global Variable  ${MAIN_THREAD}  ${thread}
-
+    Set Suite Variable  ${TEST_TOPIC}  test
+    ${thread}=  Start Consumer Threaded  topics=${TEST_TOPIC}
+    Set Suite Variable  ${MAIN_THREAD}  ${thread}
     ${producer_group_id}=  Create Producer
-    Produce  group_id=${producer_group_id}  topic=test  value=Hello
-    Produce  group_id=${producer_group_id}  topic=test  value=World
-    Produce  group_id=${producer_group_id}  topic=test  value={'test': 1}
-    Flush  ${producer_group_id}
+    Set Suite Variable  ${PRODUCER_ID}  ${producer_group_id}
+
+    ${topics}=  List Topics  ${producer_group_id}
+    ${partitions}=  Get Topic Partitions  ${topics['${TEST_TOPIC}']}
+    ${partition_id}=  Set Variable  ${partitions[0].id}
+    Set Suite Variable  ${P_ID}  ${partition_id}
+    ${tp}=  Create Topic Partition  ${TEST_TOPIC}  ${partition_id}  ${OFFSET_BEGINNING}
+
+    ${data}=  Create List  Hello  World  {'test': 1}  {'test': 2}  {'test': 3}  {'test': 4}
+    Set Suite Variable  ${TEST_DATA}  ${data}
+    Prepare Data
+
+Prepare Data
+    Produce  group_id=${PRODUCER_ID}  topic=${TEST_TOPIC}  value=Hello  partition=${P_ID}
+    Produce  group_id=${PRODUCER_ID}  topic=${TEST_TOPIC}  value=World  partition=${P_ID}
+    Produce  group_id=${PRODUCER_ID}  topic=${TEST_TOPIC}  value={'test': 1}  partition=${P_ID}
+    Produce  group_id=${PRODUCER_ID}  topic=${TEST_TOPIC}  value={'test': 2}  partition=${P_ID}
+    Produce  group_id=${PRODUCER_ID}  topic=${TEST_TOPIC}  value={'test': 3}  partition=${P_ID}
+    Produce  group_id=${PRODUCER_ID}  topic=${TEST_TOPIC}  value={'test': 4}  partition=${P_ID}
+    Wait Until Keyword Succeeds  10x  0.5s  All Messages Are Delivered  ${PRODUCER_ID}
+    Sleep  1sec  # if next command is polling messages in thread we need to wait a second
+
+All Messages Are Delivered
+    [Arguments]  ${producer_id}
+    ${count}=  Flush  ${producer_id}
+    Log  Reaming messages to be delivered: ${count}
+    Should Be Equal As Integers  ${count}  0
+
+Basic Teardown
+    [Arguments]  ${group_id}
+    Unsubscribe  ${group_id}
+    Close Consumer  ${group_id}
+
+Unassign Teardown
+    [Arguments]  ${group_id}
+    Unassign  ${group_id}
+    Close Consumer  ${group_id}
\ No newline at end of file
diff --git a/examples/test_avro.robot b/examples/test_avro.robot
new file mode 100644
index 0000000..4ed5fc2
--- /dev/null
+++ b/examples/test_avro.robot
@@ -0,0 +1,74 @@
+*** Settings ***
+Library  ConfluentKafkaLibrary
+Library  Collections
+Library  String
+
+Suite Setup  Starting Test
+
+
+*** Test Cases ***
+Avro Producer With Schemas As String Argument
+    [Setup]  Clear Messages From Thread  ${MAIN_THREAD}
+    ${value_schema}=  Set Variable  {"namespace": "example.avro","type": "record","name": "User","fields": [{"name": "name","type": "string"},{"name": "number","type": ["int","null"]}]}
+    ${key_schema}=  Set Variable  {"namespace": "example.avro","type": "record","name": "User","fields": [{"name": "name","type": "string"}]}
+    ${producer_id}=  Create Producer  schema_registry_url=http://127.0.0.1:8081
+    ...  value_schema=${value_schema}  key_schema=${key_schema}
+    ${value}=  Create Dictionary  name=Robot  number=${10}
+    Produce  group_id=${producer_id}  topic=avro_testing1  partition=${0}  value=${value}  key=${KEY}
+    Wait Until Keyword Succeeds  10x  0.5s  All Messages Are Delivered  ${producer_id}
+    Sleep  1s
+
+    ${consumer_group_id}=  Create Consumer  auto_offset_reset=earliest  schema_registry_url=http://127.0.0.1:8081
+    Subscribe Topic  group_id=${consumer_group_id}  topics=avro_testing1
+    ${messages}=  Poll  group_id=${consumer_group_id}
+    Should Be Equal  ${TEST_DATA}  ${messages}
+    ${thread_messages}=  Get Messages From Thread  ${MAIN_THREAD}
+    Should Be Equal  ${TEST_DATA}  ${thread_messages}
+    [Teardown]  Basic Teardown  ${consumer_group_id}
+
+Avro Producer With Path To Schemas
+    [Setup]  Clear Messages From Thread  ${MAIN_THREAD}
+    ${value_schema_file_path}=  Set Variable  examples/schema/producer/ValueSchema.avsc
+    ${key_schema_file_path}=  Set Variable  examples/schema/producer/KeySchema.avsc
+    ${producer_id}=  Create Producer  schema_registry_url=http://127.0.0.1:8081
+    ...  value_schema=${value_schema_file_path}  key_schema=${key_schema_file_path}
+    ${value}=  Create Dictionary  name=Robot  number=${10}
+    Produce  group_id=${producer_id}  topic=avro_testing2  partition=${0}  value=${value}  key=${KEY}
+    Wait Until Keyword Succeeds  10x  0.5s  All Messages Are Delivered  ${producer_id}
+    Sleep  1s
+
+    ${consumer_group_id}=  Create Consumer  auto_offset_reset=earliest  schema_registry_url=http://127.0.0.1:8081
+    Subscribe Topic  group_id=${consumer_group_id}  topics=avro_testing2
+    ${messages}=  Poll  group_id=${consumer_group_id}
+    Should Be Equal  ${TEST_DATA}  ${messages}
+    ${thread_messages}=  Get Messages From Thread  ${MAIN_THREAD}
+    Should Be Equal  ${TEST_DATA}  ${thread_messages}
+    [Teardown]  Basic Teardown  ${consumer_group_id}
+
+
+*** Keywords ***
+Starting Test
+    Set Suite Variable  @{TEST_TOPIC}  avro_testing1  avro_testing2
+    Set Suite Variable  &{KEY}  name=testkey
+    ${value}=  Create Dictionary  name=Robot  number=${10}
+    ${data}=  Create List  ${value}
+    Set Suite Variable  ${TEST_DATA}  ${data}
+
+    ${thread}=  Start Consumer Threaded  topics=${TEST_TOPIC}   schema_registry_url=http://127.0.0.1:8081  auto_offset_reset=latest
+    Set Suite Variable  ${MAIN_THREAD}  ${thread}
+
+All Messages Are Delivered
+    [Arguments]  ${producer_id}
+    ${count}=  Flush  ${producer_id}
+    Log  Reaming messages to be delivered: ${count}
+    Should Be Equal As Integers  ${count}  0
+
+Basic Teardown
+    [Arguments]  ${group_id}
+    Unsubscribe  ${group_id}
+    Close Consumer  ${group_id}
+
+Unassign Teardown
+    [Arguments]  ${group_id}
+    Unassign  ${group_id}
+    Close Consumer  ${group_id}
\ No newline at end of file
diff --git a/requirements.txt b/requirements.txt
index de997d1..19d202c 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,3 +1,3 @@
-robotframework==3.1.1
-confluent-kafka==1.0.0
+robotframework==3.1.2
+confluent-kafka==1.2.0
 uuid==1.30
diff --git a/setup.py b/setup.py
index cc8661f..14030e4 100644
--- a/setup.py
+++ b/setup.py
@@ -21,15 +21,14 @@
       keywords     = 'robotframework confluent kafka',
       platforms    = 'any',
       classifiers  = [
-          "Development Status :: 4 - Beta",
           "License :: OSI Approved :: Apache Software License",
           "Operating System :: OS Independent",
           "Programming Language :: Python",
           "Topic :: Software Development :: Testing"
       ],
       install_requires = [
-          'robotframework == 3.1.1',
-          'confluent-kafka[avro] == 1.2.0',
+          'robotframework == 3.1.2',
+          'confluent-kafka == 1.3.0',
           'uuid==1.30',
       ],
       package_dir  = {'' : 'src'},
diff --git a/src/ConfluentKafkaLibrary/__init__.py b/src/ConfluentKafkaLibrary/__init__.py
index 3482d06..4ca5962 100644
--- a/src/ConfluentKafkaLibrary/__init__.py
+++ b/src/ConfluentKafkaLibrary/__init__.py
@@ -1,3 +1,6 @@
+from robot.libraries.BuiltIn import BuiltIn
+import confluent_kafka
+
 from .consumer import KafkaConsumer
 from .producer import KafkaProducer
 from .version import VERSION
@@ -17,7 +20,7 @@ class ConfluentKafkaLibrary(KafkaConsumer, KafkaProducer):
     For more information about Robot Framework, see http://robotframework.org.
 
     == Examples ==
-    
+
     *Basic Consumer with predefined group_id*
 
     | ${group_id}= | `Create Consumer` | group_id=mygroup | # if group_id is not defined uuid4() is gemerated |
@@ -63,3 +66,23 @@ class ConfluentKafkaLibrary(KafkaConsumer, KafkaProducer):
     def __init__(self):
         KafkaConsumer.__init__(self)
         KafkaProducer.__init__(self)
+        BuiltIn().set_global_variable('${OFFSET_BEGINNING}', confluent_kafka.OFFSET_BEGINNING)
+        BuiltIn().set_global_variable('${OFFSET_END}', confluent_kafka.OFFSET_END)
+        BuiltIn().set_global_variable('${OFFSET_STORED}', confluent_kafka.OFFSET_STORED)
+        BuiltIn().set_global_variable('${OFFSET_INVALID}', confluent_kafka.OFFSET_INVALID)
+
+    def list_topics(self, group_id, topic=None):
+        """Request Metadata from cluster. Could be executed with consumer or producer group_id too.
+        - ``topic`` (str):  If specified, only request info about this topic, else returnfor all topics in cluster.
+        Default: `None`.
+        - ``group_id`` (str): *required* id of the created consumer or producer.
+        """
+        if group_id is None:
+            raise TypeError
+
+        if group_id in self.consumers:
+            return self.consumers[group_id].list_topics(topic).topics
+        if group_id in self.producers:
+            return self.producers[group_id].list_topics(topic).topics
+
+        raise ValueError('Consumer or producer group_id is wrong or does not exists!')
diff --git a/src/ConfluentKafkaLibrary/consumer.py b/src/ConfluentKafkaLibrary/consumer.py
index 6e3992c..f09398f 100644
--- a/src/ConfluentKafkaLibrary/consumer.py
+++ b/src/ConfluentKafkaLibrary/consumer.py
@@ -1,11 +1,7 @@
-import sys
 import uuid
-import copy
-import json
-import threading
-from threading import Thread, Timer
+from threading import Thread
+from confluent_kafka import Consumer, KafkaException, KafkaError, TopicPartition
 from confluent_kafka.avro.serializer import SerializerError
-from confluent_kafka import Consumer, KafkaError, TopicPartition
 from confluent_kafka.avro import AvroConsumer
 
 
@@ -17,47 +13,53 @@ def __init__(
         port='9092',
         topics='',
         group_id=None,
+        only_value=True,
         **kwargs
     ):
 
         super(GetMessagesThread, self).__init__()
         self.daemon = True
         self._is_running = True
+        self.only_value = only_value
         self.consumer = KafkaConsumer()
-        self.group_id = self.consumer.create_consumer(server=server, port=port, group_id=group_id, **kwargs)
+        self.group_id = self.consumer.create_consumer(group_id=group_id,
+                                                      server=server,
+                                                      port=port,
+                                                      **kwargs)
+
         if not isinstance(topics, list):
             topics = [topics]
         self.consumer.subscribe_topic(self.group_id, topics=topics)
         self.messages = []
-        self.messages += self.consumer.poll(group_id=self.group_id)
+        self.messages += self.consumer.poll(group_id=self.group_id, only_value=self.only_value)
         self.start()
 
     def run(self):
         while self._is_running:
             try:
-                self.messages += self.consumer.poll(group_id=self.group_id)
+                self.messages += self.consumer.poll(group_id=self.group_id, only_value=self.only_value)
             except RuntimeError:
+                self.consumer.unsubscribe(self.group_id)
+                self.consumer.close_consumer(self.group_id)
                 self._is_running = False
 
-    def stop(self):
-        self.consumer.unsubscribe(self.group_id)
-        self.consumer.close_consumer(self.group_id)
-        self._is_running = False
-
     def get_messages(self):
-        return self.messages
+        return self.messages[:]
+
+    def clear_messages(self):
+        self.messages.clear()
 
-class KafkaConsumer(object):
+
+class KafkaConsumer():
 
     def __init__(self):
-        self.__consumers = {}
+        self.consumers = {}
 
     def create_consumer(
         self,
         group_id=None,
         server="127.0.0.1",
         port="9092",
-        client_id='Robot',
         enable_auto_commit=True,
         auto_offset_reset="latest",
         schema_registry_url=None,
@@ -70,13 +72,8 @@ def create_consumer(
             contact to bootstrap initial cluster metadata.
             Default: `127.0.0.1`.
         - ``port`` (int): Port number. Default: `9092`.
-        - ``client_id`` (str): a name for this client. This string is passed in
-            each request to servers and can be used to identify specific
-            server-side log entries that correspond to this client. Also
-            submitted to GroupCoordinator for logging with respect to
-            consumer group administration. Default: `Robot`.
-        - ``group_id`` (str or uuid.uuid4() if not set) : name of the consumer group to join for dynamic
-            partition assignment (if enabled), and to use for fetching and
+        - ``group_id`` (str or uuid.uuid4() if not set) : name of the consumer group
+            to join for dynamic partition assignment (if enabled), and to use for fetching and
             committing offsets. If None, unique string is generated  (via uuid.uuid4())
             and offset commits are disabled. Default: `None`.
         - ``auto_offset_reset`` (str): A policy for resetting offsets on
@@ -85,7 +82,8 @@ def create_consumer(
             other value will raise the exception. Default: `latest`.
         - ``enable_auto_commit`` (bool): If true the consumer's offset will be
             periodically committed in the background. Default: `True`.
-        - ``schema_registry_url`` (str): *required* for Avro Consumer. Full URL to avro schema endpoint.
+        - ``schema_registry_url`` (str): *required* for Avro Consumer.
+            Full URL to avro schema endpoint.
 
         Note:
         Configuration parameters are described in more detail at
@@ -114,68 +112,130 @@ def create_consumer(
                 },
                 **kwargs})
 
-        self.__consumers[group_id] = consumer
+        self.consumers[group_id] = consumer
         return group_id
 
-    def _is_assigned(self, group_id, topic_partition):
-        for tp in topic_partition:
-            if tp in self.__consumers[group_id].assignment():
-                return True
-        return False
+    def create_topic_partition(self, topic_name, partition=None, offset=None):
+        """Returns TopicPartiton object based on
+           https://docs.confluent.io/current/clients/confluent-kafka-python/#topicpartition
 
-    def assign_to_topic_partition(self, group_id, topic_partition):
-        """Assign a list of TopicPartitions.
-
-        - ``partitions`` (list of `TopicPartition`): Assignment for this instance.
+            - ``topic_name`` (str): Topic name.
+            - ``partition`` (int): Partition id.
+            - ``offset`` (int): Initial partition offset.
         """
+        if partition is not None and offset is not None:
+            return TopicPartition(topic_name, partition, offset)
+        elif partition is None:
+            return TopicPartition(topic_name, offset)
+        elif offset is None:
+            return TopicPartition(topic_name, partition)
+        else:
+            return TopicPartition(topic_name)
 
-        if isinstance(topic_partition, TopicPartition):
-            topic_partition = [topic_partition]
-        if not self._is_assigned(group_id, topic_partition):
-            self.__consumers[group_id].assign(topic_partition)
+    def get_topic_partitions(self, topic):
+        """Returns dictionary of all TopicPartitons in topic (topic.partitions).
+        """
+        return topic.partitions
 
-    def subscribe_topic(self, group_id, topics):
+    def subscribe_topic(self, group_id, topics, **kwargs):
         """Subscribe to a list of topics, or a topic regex pattern.
+           https://docs.confluent.io/current/clients/confluent-kafka-python/#confluent_kafka.Consumer.subscribe
 
         - ``topics`` (list): List of topics for subscription.
         """
-
         if not isinstance(topics, list):
             topics = [topics]
-        self.__consumers[group_id].subscribe(topics)
+        self.consumers[group_id].subscribe(topics, **kwargs)
+
+    def get_watermark_offsets(self, group_id, topic_partition, **kwargs):
+        """Retrieve low and high offsets for partition.
+        """
+        if not isinstance(topic_partition, TopicPartition):
+            raise TypeError('topic_partition needs to be TopicPartition() type!')
+        return self.consumers[group_id].get_watermark_offsets(topic_partition, **kwargs)
+
+    def get_assignment(self, group_id):
+        return self.consumers[group_id].assignment()
+
+    def assign_to_topic_partition(self, group_id, topic_partitions):
+        """Assign a list of TopicPartitions.
+
+        - ``topic_partitions`` (`TopicPartition` or list of `TopicPartition`): Assignment for this instance.
+        """
+        if isinstance(topic_partitions, TopicPartition):
+            topic_partitions = [topic_partitions]
+        for topic_partition in topic_partitions:
+            if topic_partition not in self.consumers[group_id].assignment():
+                self.consumers[group_id].assign(topic_partitions)
+
+    def unassign(self, group_id):
+        self.consumers[group_id].unassign()
 
     def unsubscribe(self, group_id):
         """Unsubscribe of topics.
         """
-        self.__consumers[group_id].unsubscribe()
+        self.consumers[group_id].unsubscribe()
 
     def close_consumer(self, group_id):
         """Close down and terminate the Kafka Consumer.
         """
-        self.__consumers[group_id].close()
+        self.consumers[group_id].close()
+
+    def seek(self, group_id, topic_partition):
+        """https://docs.confluent.io/current/clients/confluent-kafka-python/#confluent_kafka.Consumer.seek
+        """
+        return self.consumers[group_id].seek(topic_partition)
+
+    def get_position(self, group_id, topic_partitions):
+        """Retrieve current positions (offsets) for the list of partitions.
+
+            - ``topic_partitions`` (`TopicPartition` or list of `TopicPartition`): Assignment for this instance.
+        """
+        if isinstance(topic_partitions, TopicPartition):
+            topic_partitions = [topic_partitions]
+        return self.consumers[group_id].position(topic_partitions)
+
+    def pause(self, group_id, topic_partitions):
+        """Pause consumption for the provided list of partitions.
+        """
+        if isinstance(topic_partitions, TopicPartition):
+            topic_partitions = [topic_partitions]
+        self.consumers[group_id].pause(topic_partitions)
+
+    def resume(self, group_id, topic_partitions):
+        """Resume consumption for the provided list of partitions.
+        """
+        if isinstance(topic_partitions, TopicPartition):
+            topic_partitions = [topic_partitions]
+        self.consumers[group_id].resume(topic_partitions)
 
     def poll(
         self,
         group_id,
         timeout=1,
         max_records=1,
-        poll_attempts=10
+        poll_attempts=10,
+        only_value=True,
+        decode_format=None
     ):
         """Fetch and return messages from assigned topics / partitions as list.
+        - ``timeout`` (int): Seconds spent waiting in poll if data is not available in the buffer.\n
         - ``max_records`` (int): maximum number of messages to get from poll. Default: 1.
-        - ``timeout`` (int): Seconds spent waiting in poll if data is not available in the buffer.
-        If 0, returns immediately with any records that are available currently in the buffer, else returns empty.
-        Must not be negative. Default: `1`
-        -  ``poll_attempts`` (int): Attempts to consume messages and endless looping prevention.
+        If 0, returns immediately with any records that are available currently in the buffer,
+        else returns empty. Must not be negative. Default: `1`
+        - ``poll_attempts`` (int): Attempts to consume messages and endless looping prevention.
         Sometimes the first messages are None or the topic could be empty. Default: `10`.
+        - ``only_value`` (bool): Return only message.value(). Default: `True`.
+        - ``decode_format`` (str) - If you need to decode data to specific format
+            (See https://docs.python.org/3/library/codecs.html#standard-encodings). Default: None.
         """
 
         messages = []
         while poll_attempts > 0:
             try:
-                msg = self.__consumers[group_id].poll(timeout=timeout)
-            except SerializerError as e:
-                print('Message deserialization failed for {}: {}'.format(msg, e))
+                msg = self.consumers[group_id].poll(timeout=timeout)
+            except SerializerError as err:
+                print('Message deserialization failed for {}: {}'.format(msg, err))
                 break
 
             if msg is None:
@@ -183,30 +243,24 @@ def poll(
                 continue
 
             if msg.error():
-                if msg.error().code() == KafkaError._PARTITION_EOF:
-                    poll_attempts = 0
-                    continue
-                else:
-                    print(msg.error())
-                    break
+                raise KafkaException(msg.error())
 
-            messages.append(msg.value())
+            if only_value:
+                messages.append(msg.value())
+            else:
+                messages.append(msg)
 
             if len(messages) == max_records:
-                return messages
+                break
+
+        if decode_format:
+            messages = self._decode_data(data=messages, decode_format=decode_format)
 
         return messages
 
-    # Experimental keywords
-    def decode_data(self, data, decode_format, remove_zero_bytes=False):
-        """ TODO
-        """
-        if decode_format and remove_zero_bytes:
-            return [record.decode(str(decode_format)).replace('\x00', '') for record in data]
-        elif decode_format and not remove_zero_bytes:
+    def _decode_data(self, data, decode_format):
+        if decode_format:
             return [record.decode(str(decode_format)) for record in data]
-        elif not decode_format and remove_zero_bytes:
-            return [record.replace('\x00', '') for record in data]
         else:
             return data
 
@@ -217,30 +271,48 @@ def start_consumer_threaded(
         group_id=None,
         server='127.0.0.1',
         port='9092',
+        only_value=True,
         **kwargs
     ):
+        """Run consumer in daemon thread and store data from topics. To read and work with this
+           collected data use keyword `Get Messages From Thread`.
+           Could be used at the Test setup or in each test.
+           This is useful when you are reading always the same topics and you don't want to create
+           consumer in each test to poll data. You can create as many consumers in the Test setup
+           as you want and then in test just read data with `Get Messages From Thread` keyword.
+        - ``topics`` (list): List of topics for subscription.
+        - ``group_id`` (str or uuid.uuid4() if not set) : name of the consumer group to join for
+            dynamic partition assignment (if enabled), and to use for fetching and
+            committing offsets. If None, unique string is generated  (via uuid.uuid4())
+            and offset commits are disabled. Default: `None`.
+        """
         if group_id is None:
             group_id = str(uuid.uuid4())
         if topics is None:
             raise ValueError("Topics can not be empty!")
 
-        t = GetMessagesThread(server, port, topics, group_id=group_id, **kwargs)
-        return t
+        consumer_thread = GetMessagesThread(server, port, topics, group_id=group_id, only_value=only_value, **kwargs)
+        return consumer_thread
 
-    def get_messages_from_thread(self, running_thread, decode_format=None, remove_zero_bytes=False):
-        """ Returns all records gathered from specific thread
-            - ``running_thread`` (Thread object) - thread which was executed by `Start Consumer Threaded`
-            - ``decode_data`` (str) - If you need to decode data to specific format
-                (See https://docs.python.org/3/library/codecs.html#standard-encodings). Default: None.
-            - ``remove_zero_bytes`` (bool) - When you are working with byte streams
-                you can end up with a lot of '\\x00' bytes you want to remove. Default: False.
+    def get_messages_from_thread(self, running_thread, decode_format=None):
+        """Returns all records gathered from specific thread
+        - ``running_thread`` (Thread object) - thread which was executed with
+            `Start Consumer Threaded` keyword
+        - ``decode_format`` (str) - If you need to decode data to specific format
+            (See https://docs.python.org/3/library/codecs.html#standard-encodings). Default: None.
         """
-        records = self.decode_data(
+        records = self._decode_data(
             data=running_thread.get_messages(),
-            decode_format=decode_format,
-            remove_zero_bytes=remove_zero_bytes
+            decode_format=decode_format
         )
         return records
 
-    def stop_thread(self, running_thread):
-        running_thread.stop()
+    def clear_messages_from_thread(self, running_thread):
+        """Remove all records gathered from specific thread
+        - ``running_thread`` (Thread object) - thread which was executed with
+            `Start Consumer Threaded` keyword
+        """
+        try:
+            running_thread.clear_messages()
+        except:
+            print('Messages was not removed from thread %s!', running_thread)
diff --git a/src/ConfluentKafkaLibrary/producer.py b/src/ConfluentKafkaLibrary/producer.py
index 9a9a1cb..b7a3393 100644
--- a/src/ConfluentKafkaLibrary/producer.py
+++ b/src/ConfluentKafkaLibrary/producer.py
@@ -1,17 +1,29 @@
 import sys
 import uuid
 import copy
+import os
 import json
 from confluent_kafka.avro.serializer import SerializerError
 from confluent_kafka import Producer, KafkaError, TopicPartition
 from confluent_kafka import avro
+from avro import schema
 from confluent_kafka.avro import AvroProducer
 
 
 class KafkaProducer(object):
 
     def __init__(self):
-        self.__producers = {}
+        self.producers = {}
+
+    def load_schema(self, data):
+        if type(data) == schema.RecordSchema:
+            data = data
+        elif os.path.exists(data):
+            data = avro.load(data)
+        elif type(data) is str:
+            data = str(data)
+            data = avro.loads(data)
+        return data
 
     def create_producer(
         self,
@@ -20,8 +32,8 @@ def create_producer(
         client_id='Robot',
         group_id=None,
         schema_registry_url=None,
-        value_schema_str=None,
-        key_schema_str=None,
+        value_schema=None,
+        key_schema=None,
         **kwargs
     ):
         """Create Kafka Producer and returns its `group_id` as string.
@@ -43,16 +55,18 @@ def create_producer(
             committing offsets. If None, unique string is generated  (via uuid.uuid4())
             and offset commits are disabled. Default: `None`.
         - ``schema_registry_url`` (str): *required* for Avro Consumer. Full URL to avro schema endpoint.
-        - ``value_schema_str`` (str): Optional default avro schema for value. Default: `None`
-        - ``key_schema_str`` (str): Optional default avro schema for key. Default: `None`
+        - ``value_schema`` (str): Optional default avro schema for value or path to file with schema. Default: `None`
+        - ``key_schema`` (str): Optional default avro schema for key. Default: `None`
 
         """
         if group_id is None:
             group_id = str(uuid.uuid4())
 
         if schema_registry_url:
-            value_schema = avro.loads(value_schema_str)
-            key_schema = avro.loads(key_schema_str)
+            if value_schema:
+                value_schema = self.load_schema(value_schema)
+            if key_schema:
+                key_schema = self.load_schema(key_schema)
 
             producer = AvroProducer({
                 'bootstrap.servers': '{}:{}'.format(server, port),
@@ -68,40 +82,36 @@ def create_producer(
                 'group.id': group_id,
                 **kwargs})
 
-        self.__producers[group_id] = producer
+        self.producers[group_id] = producer
         return group_id
 
     def produce(
         self,
         group_id,
         topic,
-        value,
+        value=None,
         key=None,
-        value_encoding='utf-8',
-        callback_func=None
+        **kwargs
     ):
-        """Produce message to topic asynchronously to Kafka by encoding with specified or default avro schema.
+        """Produce message to topic asynchronously to Kafka by encoding with specified or default avro schema.\n
+        https://docs.confluent.io/current/clients/confluent-kafka-python/#confluent_kafka.Producer.produce
 
-        - ``topic` (str) : name of the topic where to produce message.
-        - ``data` (str) : Message which produce to topic.
-        - ``data_encoding` (str) : encode to specific format. Default: `utf-8`.
-        - ``callback_func` (func) : the function that will be called from poll()
-        - ``key`` (object) : An object to serialize. Default: `None`.
-            when the message has been successfully delivered or permanently fails delivery..
+        - ``topic`` (str) : name of the topic where to produce message.
+        - ``value`` (str|bytes): Message payload.
+        - ``key`` (str|bytes): Message key. Default: `None`.
+        - ``partition`` (int): Partition to produce to, else uses the configured built-in partitioner.
         """
-        self.__producers[group_id].produce(
+        self.producers[group_id].produce(
             topic=topic,
-            value=value.encode(value_encoding),
-            callback=callback_func,
-            key=key
+            value=value,
+            key=key,
+            **kwargs
         )
 
-    def flush(self, group_id, timeout=1.0):
-        """Wait for all messages in the Producer queue to be delivered.
+    def flush(self, group_id, timeout=0.1):
+        """Wait for all messages in the Producer queue to be delivered. Returns the number of messages still in queue.
         This is a convenience method that calls poll() until len() is zero or the optional timeout elapses.
-        - `timeout` (real) : Optional timeout. Default: `1.0`.
+        - `timeout` (real) : Optional timeout. Default: `0.1`.
         """
-        self.__producers[group_id].flush(timeout)
-
-    def list_topics(self, group_id,):
-        return self.__producers[group_id].list_topics()
+        messages_in_queue = self.producers[group_id].flush(timeout)
+        return messages_in_queue
diff --git a/src/ConfluentKafkaLibrary/version.py b/src/ConfluentKafkaLibrary/version.py
index d325e71..9e23359 100644
--- a/src/ConfluentKafkaLibrary/version.py
+++ b/src/ConfluentKafkaLibrary/version.py
@@ -1 +1 @@
-VERSION = '0.3.0'
+VERSION = '1.3.0'