Repository: robooo/robotframework-ConfluentKafkaLibrary (language: python)

Issue:
Implement missing AdminClient
Implement Kafka admin client to be able to create, view, alter, and delete topics and resources

Implement missing AdminClient
Implement Kafka admin client to be able to create, view, alter, and delete topics and resources

Implement missing AdminClient
Implement Kafka admin client to be able to create, view, alter, and delete topics and resources

Implement missing AdminClient
Implement Kafka admin client to be able to create, view, alter, and delete topics and resources

Implement missing AdminClient
Implement Kafka admin client to be able to create, view, alter, and delete topics and resources

Implement missing AdminClient
Implement Kafka admin client to be able to create, view, alter, and delete topics and resources

Implement missing AdminClient
Implement Kafka admin client to be able to create, view, alter, and delete topics and resources

Implement missing AdminClient
Implement Kafka admin client to be able to create, view, alter, and delete topics and resources



Discussion / hints:

Fix patch:
diff --git a/docs/index.html b/docs/index.html
index 8fb1298..28bd9fa 100644
--- a/docs/index.html
+++ b/docs/index.html
@@ -1102,7 +1102,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 = {"name": "ConfluentKafkaLibrary", "doc": "<p>ConfluentKafkaLibrary is a Robot Framework library which wraps up <a href=\"https://github.com/confluentinc/confluent-kafka-python\">confluent-kafka-python</a>. Library supports more functionality like running more clients based on <span class=\"name\">group_id</span> or running them in threaded mode during the tests, decoding of gathered data etc. (<span class=\"name\">See</span> <a href=\"#Examples\" class=\"name\">Examples</a>).</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</a>. For more information about Robot Framework, see <a href=\"http://robotframework.org\">http://robotframework.org</a>.</p>\n<h3 id=\"Examples\">Examples</h3>\n<p>See <a href=\"https://github.com/robooo/robotframework-ConfluentKafkaLibrary/tree/master/examples\">repo examples</a>.</p>\n<p><b>Basic Consumer with predefined group_id</b></p>\n<table border=\"1\">\n<tr>\n<td>${group_id}=</td>\n<td><a href=\"#Create%20Consumer\" class=\"name\">Create Consumer</a></td>\n<td>group_id=mygroup</td>\n<td># if group_id is not defined uuid4() is gemerated</td>\n</tr>\n<tr>\n<td><a href=\"#Subscribe%20Topic\" class=\"name\">Subscribe Topic</a></td>\n<td>group_id=${group_id}</td>\n<td>topics=test_topic</td>\n<td></td>\n</tr>\n<tr>\n<td>${result}=</td>\n<td><a href=\"#Poll\" class=\"name\">Poll</a></td>\n<td>group_id=${group_id}</td>\n<td>max_records=5</td>\n</tr>\n<tr>\n<td><span class=\"name\">Log</span></td>\n<td>${result}</td>\n<td></td>\n<td></td>\n</tr>\n<tr>\n<td><a href=\"#Unsubscribe\" class=\"name\">Unsubscribe</a></td>\n<td>${group_id}</td>\n<td></td>\n<td></td>\n</tr>\n<tr>\n<td><a href=\"#Close%20Consumer\" class=\"name\">Close Consumer</a></td>\n<td>${group_id}</td>\n<td></td>\n<td></td>\n</tr>\n</table>\n<p><b>More Consumers</b></p>\n<table border=\"1\">\n<tr>\n<td>${group_id_1}=</td>\n<td><a href=\"#Create%20Consumer\" class=\"name\">Create Consumer</a></td>\n<td></td>\n<td></td>\n</tr>\n<tr>\n<td><a href=\"#Subscribe%20Topic\" class=\"name\">Subscribe Topic</a></td>\n<td>group_id=${group_id_1}</td>\n<td>topics=topic1</td>\n<td></td>\n</tr>\n<tr>\n<td>${group_id_2}=</td>\n<td><a href=\"#Create%20Consumer\" class=\"name\">Create Consumer</a></td>\n<td></td>\n<td></td>\n</tr>\n<tr>\n<td><a href=\"#Subscribe%20Topic\" class=\"name\">Subscribe Topic</a></td>\n<td>group_id=${group_id_2}</td>\n<td>topics=topic2</td>\n<td></td>\n</tr>\n<tr>\n<td>${result_1}=</td>\n<td><a href=\"#Poll\" class=\"name\">Poll</a></td>\n<td>group_id=${group_id_1}</td>\n<td>max_records=5</td>\n</tr>\n<tr>\n<td>${result_2}=</td>\n<td><a href=\"#Poll\" class=\"name\">Poll</a></td>\n<td>group_id=${group_id_2}</td>\n<td>max_records=2</td>\n</tr>\n<tr>\n<td><a href=\"#Unsubscribe\" class=\"name\">Unsubscribe</a></td>\n<td>${group_id_1}</td>\n<td></td>\n<td></td>\n</tr>\n<tr>\n<td><a href=\"#Unsubscribe\" class=\"name\">Unsubscribe</a></td>\n<td>${group_id_2}</td>\n<td></td>\n<td></td>\n</tr>\n<tr>\n<td><a href=\"#Close%20Consumer\" class=\"name\">Close Consumer</a></td>\n<td>${group_id_1}</td>\n<td></td>\n<td></td>\n</tr>\n<tr>\n<td><a href=\"#Close%20Consumer\" class=\"name\">Close Consumer</a></td>\n<td>${group_id_2}</td>\n<td></td>\n<td></td>\n</tr>\n</table>\n<p><b>Run Avro Consumer over HTTPS and threaded:</b></p>\n<table border=\"1\">\n<tr>\n<td>${thread}=</td>\n<td><a href=\"#Start%20Consumer%20Threaded\" class=\"name\">Start Consumer Threaded</a></td>\n<td></td>\n</tr>\n<tr>\n<td></td>\n<td>...  topics=test_topic</td>\n<td></td>\n</tr>\n<tr>\n<td></td>\n<td>...  schema_registry_url=https://localhost:8081</td>\n<td></td>\n</tr>\n<tr>\n<td></td>\n<td>...  auto_offset_reset=earliest</td>\n<td># We want to get all messages</td>\n</tr>\n<tr>\n<td></td>\n<td>...  security.protocol=ssl</td>\n<td></td>\n</tr>\n<tr>\n<td></td>\n<td>...  schema.registry.ssl.ca.location=/home/user/cert/caproxy.pem</td>\n<td></td>\n</tr>\n<tr>\n<td></td>\n<td>...  ssl.ca.location=/home/user/cert/caproxy.pem</td>\n<td></td>\n</tr>\n<tr>\n<td></td>\n<td>...  ssl.certificate.location=/home/user/cert/kafka-client-cert.pem</td>\n<td></td>\n</tr>\n<tr>\n<td></td>\n<td>...  ssl.key.location=/home/user/cert/kafka-client-key.pem</td>\n<td></td>\n</tr>\n<tr>\n<td><span class=\"name\">Log</span></td>\n<td><span class=\"name\">Execute commands which should push some data to topic</span></td>\n<td></td>\n</tr>\n<tr>\n<td>${messages}=</td>\n<td><a href=\"#Get%20Messages%20From%20Thread\" class=\"name\">Get Messages From Thread</a></td>\n<td>${thread}</td>\n</tr>\n</table>", "version": "1.7.0", "generated": "2021-10-27 20:13:38", "type": "LIBRARY", "scope": "GLOBAL", "docFormat": "HTML", "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/__init__.py", "lineno": 12, "tags": [], "inits": [], "keywords": [{"name": "Assign To Topic Partition", "args": [{"name": "group_id", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "group_id"}, {"name": "topic_partitions", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "topic_partitions"}], "doc": "<p>Assign a list of TopicPartitions.</p>\n<ul>\n<li><code>topic_partitions</code> (<span class=\"name\">TopicPartition</span> or list of <span class=\"name\">TopicPartition</span>): Assignment for this instance.</li>\n</ul>", "shortdoc": "Assign a list of TopicPartitions.", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/consumer.py", "lineno": 182}, {"name": "Clear Messages From Thread", "args": [{"name": "running_thread", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "running_thread"}], "doc": "<p>Remove all records gathered from specific thread</p>\n<ul>\n<li><code>running_thread</code> (Thread object) - thread which was executed with <a href=\"#Start%20Consumer%20Threaded\" class=\"name\">Start Consumer Threaded</a> keyword</li>\n</ul>", "shortdoc": "Remove all records gathered from specific thread - ``running_thread`` (Thread object) - thread which was executed with     `Start Consumer Threaded` keyword", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/consumer.py", "lineno": 349}, {"name": "Close Consumer", "args": [{"name": "group_id", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "group_id"}], "doc": "<p>Close down and terminate the Kafka Consumer.</p>", "shortdoc": "Close down and terminate the Kafka Consumer.", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/consumer.py", "lineno": 201}, {"name": "Create Consumer", "args": [{"name": "group_id", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "group_id=None"}, {"name": "server", "types": [], "defaultValue": "127.0.0.1", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "server=127.0.0.1"}, {"name": "port", "types": [], "defaultValue": "9092", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "port=9092"}, {"name": "enable_auto_commit", "types": [], "defaultValue": "True", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "enable_auto_commit=True"}, {"name": "auto_offset_reset", "types": [], "defaultValue": "latest", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "auto_offset_reset=latest"}, {"name": "schema_registry_url", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "schema_registry_url=None"}, {"name": "auto_create_topics", "types": [], "defaultValue": "True", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "auto_create_topics=True"}, {"name": "key_deserializer", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "key_deserializer=None"}, {"name": "value_deserializer", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "value_deserializer=None"}, {"name": "legacy", "types": [], "defaultValue": "True", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "legacy=True"}, {"name": "kwargs", "types": [], "defaultValue": null, "kind": "VAR_NAMED", "required": false, "repr": "**kwargs"}], "doc": "<p>Create Kafka Consumer and returns its <span class=\"name\">group_id</span> as string.</p>\n<p>Keyword Arguments:</p>\n<ul>\n<li><code>server</code>: (str): IP address / domain, that the consumer should contact to bootstrap initial cluster metadata. Default: <span class=\"name\">127.0.0.1</span>.</li>\n<li><code>port</code> (int): Port number. Default: <span class=\"name\">9092</span>.</li>\n<li><code>group_id</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</span>.</li>\n<li><code>auto_offset_reset</code> (str): A policy for resetting offsets on OffsetOutOfRange errors: <span class=\"name\">earliest</span> will move to the oldest available message, <span class=\"name\">latest</span> will move to the most recent. Any other value will raise the exception. Default: <span class=\"name\">latest</span>.</li>\n<li><code>enable_auto_commit</code> (bool): If true the consumer's offset will be periodically committed in the background. Default: <span class=\"name\">True</span>.</li>\n<li><code>schema_registry_url</code> (str): <b>required</b> for Avro Consumer. Full URL to avro schema endpoint.</li>\n<li><code>auto_create_topics</code> (bool): Consumers no longer trigger auto creation of topics, will be removed in future release. Default: <span class=\"name\">True</span>.</li>\n<li><code>legacy</code> (bool): Activate SerializingConsumer if 'False' else AvroConsumer (legacy) is used. Will be removed when confluent-kafka will deprecate this. Default: <span class=\"name\">True</span>.</li>\n</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</a></p>", "shortdoc": "Create Kafka Consumer and returns its `group_id` as string.", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/consumer.py", "lineno": 59}, {"name": "Create Producer", "args": [{"name": "server", "types": [], "defaultValue": "127.0.0.1", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "server=127.0.0.1"}, {"name": "port", "types": [], "defaultValue": "9092", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "port=9092"}, {"name": "client_id", "types": [], "defaultValue": "Robot", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "client_id=Robot"}, {"name": "group_id", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "group_id=None"}, {"name": "schema_registry_url", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "schema_registry_url=None"}, {"name": "value_schema", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "value_schema=None"}, {"name": "key_schema", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "key_schema=None"}, {"name": "key_serializer", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "key_serializer=None"}, {"name": "value_serializer", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "value_serializer=None"}, {"name": "legacy", "types": [], "defaultValue": "True", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "legacy=True"}, {"name": "kwargs", "types": [], "defaultValue": null, "kind": "VAR_NAMED", "required": false, "repr": "**kwargs"}], "doc": "<p>Create Kafka Producer and returns its <span class=\"name\">group_id</span> as string. If <span class=\"name\">schema_registry_url</span> is used, Kafka Producer client which does avro schema encoding to messages is created instead.</p>\n<p>Keyword Arguments:</p>\n<ul>\n<li><code>server</code>: (str): IP address / domain, that the consumer should contact to bootstrap initial cluster metadata. Default: <span class=\"name\">127.0.0.1</span>.</li>\n<li><code>port</code> (int): Port number. Default: <span class=\"name\">9092</span>.</li>\n<li><code>client_id</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</span>.</li>\n<li><code>schema_registry_url</code> (str): <b>required</b> for Avro Consumer. Full URL to avro schema endpoint.</li>\n<li><code>value_schema</code> (str): Optional default avro schema for value or path to file with schema. Default: <span class=\"name\">None</span></li>\n<li><code>key_schema</code> (str): Optional default avro schema for key. Default: <span class=\"name\">None</span></li>\n<li><code>legacy</code> (bool): Activate SerializingProducer if 'False' else AvroProducer (legacy) is used. Will be removed when confluent-kafka will deprecate this. Default: <span class=\"name\">True</span>.</li>\n</ul>", "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": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/producer.py", "lineno": 29}, {"name": "Create Topic Partition", "args": [{"name": "topic_name", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "topic_name"}, {"name": "partition", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "partition=None"}, {"name": "offset", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "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</a></p>\n<ul>\n<li><code>topic_name</code> (str): Topic name.</li>\n<li><code>partition</code> (int): Partition id.</li>\n<li><code>offset</code> (int): Initial partition offset.</li>\n</ul>", "shortdoc": "Returns TopicPartiton object based on https://docs.confluent.io/current/clients/confluent-kafka-python/#topicpartition", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/consumer.py", "lineno": 140}, {"name": "Flush", "args": [{"name": "group_id", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "group_id"}, {"name": "timeout", "types": [], "defaultValue": "0.1", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "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.</p>\n<ul>\n<li><span class=\"name\">timeout</span> (real) : Optional timeout. Default: <span class=\"name\">0.1</span>.</li>\n</ul>", "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": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/producer.py", "lineno": 119}, {"name": "Get All Consumers", "args": [], "doc": "<p>Returns all non-threaded consumers</p>", "shortdoc": "Returns all non-threaded consumers", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/consumer.py", "lineno": 135}, {"name": "Get Assignment", "args": [{"name": "group_id", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "group_id"}], "doc": "", "shortdoc": "", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/consumer.py", "lineno": 179}, {"name": "Get Avro Deserializer", "args": [{"name": "schema_str", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "schema_str"}, {"name": "schema_registry_client", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "schema_registry_client"}, {"name": "from_dict", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "from_dict=None"}], "doc": "", "shortdoc": "", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/serialization.py", "lineno": 28}, {"name": "Get Avro Serializer", "args": [{"name": "schema_str", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "schema_str"}, {"name": "schema_registry_client", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "schema_registry_client"}, {"name": "to_dict", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "to_dict=None"}, {"name": "conf", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "conf=None"}], "doc": "", "shortdoc": "", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/serialization.py", "lineno": 7}, {"name": "Get Double Deserializer", "args": [], "doc": "", "shortdoc": "", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/serialization.py", "lineno": 31}, {"name": "Get Double Serializer", "args": [], "doc": "", "shortdoc": "", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/serialization.py", "lineno": 10}, {"name": "Get Integer Deserializer", "args": [], "doc": "", "shortdoc": "", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/serialization.py", "lineno": 34}, {"name": "Get Integer Serializer", "args": [], "doc": "", "shortdoc": "", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/serialization.py", "lineno": 13}, {"name": "Get Json Deserializer", "args": [{"name": "schema_str", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "schema_str"}, {"name": "from_dict", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "from_dict=None"}], "doc": "", "shortdoc": "", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/serialization.py", "lineno": 37}, {"name": "Get Json Serializer", "args": [{"name": "schema_str", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "schema_str"}, {"name": "schema_registry_client", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "schema_registry_client"}, {"name": "to_dict", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "to_dict=None"}, {"name": "conf", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "conf=None"}], "doc": "", "shortdoc": "", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/serialization.py", "lineno": 16}, {"name": "Get Messages From Thread", "args": [{"name": "running_thread", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "running_thread"}, {"name": "decode_format", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "decode_format=None"}], "doc": "<p>Returns all records gathered from specific thread</p>\n<ul>\n<li><code>running_thread</code> (Thread object) - thread which was executed with <a href=\"#Start%20Consumer%20Threaded\" class=\"name\">Start Consumer Threaded</a> keyword</li>\n<li><code>decode_format</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</a>). Default: None.</li>\n</ul>", "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": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/consumer.py", "lineno": 336}, {"name": "Get Position", "args": [{"name": "group_id", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "group_id"}, {"name": "topic_partitions", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "topic_partitions"}], "doc": "<p>Retrieve current positions (offsets) for the list of partitions.</p>\n<ul>\n<li><code>topic_partitions</code> (<span class=\"name\">TopicPartition</span> or list of <span class=\"name\">TopicPartition</span>): Assignment for this instance.</li>\n</ul>", "shortdoc": "Retrieve current positions (offsets) for the list of partitions.", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/consumer.py", "lineno": 212}, {"name": "Get Protobuf Deserializer", "args": [{"name": "message_type", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "message_type"}], "doc": "", "shortdoc": "", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/serialization.py", "lineno": 40}, {"name": "Get Protobuf Serializer", "args": [{"name": "msg_type", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "msg_type"}, {"name": "schema_registry_client", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "schema_registry_client"}, {"name": "conf", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "conf=None"}], "doc": "", "shortdoc": "", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/serialization.py", "lineno": 19}, {"name": "Get Schema Registry Client", "args": [{"name": "conf", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "conf"}], "doc": "", "shortdoc": "", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/__init__.py", "lineno": 97}, {"name": "Get String Deserializer", "args": [{"name": "codec", "types": [], "defaultValue": "utf_8", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "codec=utf_8"}], "doc": "", "shortdoc": "", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/serialization.py", "lineno": 43}, {"name": "Get String Serializer", "args": [{"name": "codec", "types": [], "defaultValue": "utf_8", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "codec=utf_8"}], "doc": "", "shortdoc": "", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/serialization.py", "lineno": 22}, {"name": "Get Topic Partitions", "args": [{"name": "topic", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "topic"}], "doc": "<p>Returns dictionary of all TopicPartitons in topic (topic.partitions).</p>", "shortdoc": "Returns dictionary of all TopicPartitons in topic (topic.partitions).", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/consumer.py", "lineno": 157}, {"name": "Get Watermark Offsets", "args": [{"name": "group_id", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "group_id"}, {"name": "topic_partition", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "topic_partition"}, {"name": "kwargs", "types": [], "defaultValue": null, "kind": "VAR_NAMED", "required": false, "repr": "**kwargs"}], "doc": "<p>Retrieve low and high offsets for partition.</p>", "shortdoc": "Retrieve low and high offsets for partition.", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/consumer.py", "lineno": 172}, {"name": "List Topics", "args": [{"name": "group_id", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "group_id"}, {"name": "topic", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "topic=None"}], "doc": "<p>Request Metadata from cluster. Could be executed with consumer or producer group_id too.</p>\n<ul>\n<li><code>topic</code> (str):  If specified, only request info about this topic, else returnfor all topics in cluster.</li>\n</ul>\n<p>Default: <span class=\"name\">None</span>.</p>\n<ul>\n<li><code>group_id</code> (str): <b>required</b> id of the created consumer or producer.</li>\n</ul>", "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": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/__init__.py", "lineno": 81}, {"name": "Load Schema", "args": [{"name": "data", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "data"}], "doc": "", "shortdoc": "", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/producer.py", "lineno": 19}, {"name": "Pause", "args": [{"name": "group_id", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "group_id"}, {"name": "topic_partitions", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "topic_partitions"}], "doc": "<p>Pause consumption for the provided list of partitions.</p>", "shortdoc": "Pause consumption for the provided list of partitions.", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/consumer.py", "lineno": 221}, {"name": "Poll", "args": [{"name": "group_id", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "group_id"}, {"name": "timeout", "types": [], "defaultValue": "1", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "timeout=1"}, {"name": "max_records", "types": [], "defaultValue": "1", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "max_records=1"}, {"name": "poll_attempts", "types": [], "defaultValue": "10", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "poll_attempts=10"}, {"name": "only_value", "types": [], "defaultValue": "True", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "only_value=True"}, {"name": "auto_create_topics", "types": [], "defaultValue": "True", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "auto_create_topics=True"}, {"name": "decode_format", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "decode_format=None"}, {"name": "fail_on_deserialization", "types": [], "defaultValue": "False", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "fail_on_deserialization=False"}], "doc": "<p>Fetch and return messages from assigned topics / partitions as list.</p>\n<ul>\n<li><code>timeout</code> (int): Seconds spent waiting in poll if data is not available in the buffer.</li>\n</ul>\n<ul>\n<li><code>max_records</code> (int): maximum number of messages to get from poll. Default: 1.</li>\n</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</span></p>\n<ul>\n<li><code>poll_attempts</code> (int): Attempts to consume messages and endless looping prevention.</li>\n</ul>\n<p>Sometimes the first messages are None or the topic could be empty. Default: <span class=\"name\">10</span>.</p>\n<ul>\n<li><code>only_value</code> (bool): Return only message.value(). Default: <span class=\"name\">True</span>.</li>\n<li><code>decode_format</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</a>). Default: None.</li>\n<li><code>auto_create_topics</code> (bool): Consumers no longer trigger auto creation of topics, will be removed in future release. If True then the error message UNKNOWN_TOPIC_OR_PART is ignored. Default: <span class=\"name\">True</span>.</li>\n<li><code>fail_on_deserialization</code> (bool): If True and message deserialization fails, will raise a SerializerError exception; on False will just stop the current poll and return the message so far. Default: <span class=\"name\">False</span>.</li>\n</ul>", "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": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/consumer.py", "lineno": 235}, {"name": "Produce", "args": [{"name": "group_id", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "group_id"}, {"name": "topic", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "topic"}, {"name": "value", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "value=None"}, {"name": "key", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "key=None"}, {"name": "kwargs", "types": [], "defaultValue": null, "kind": "VAR_NAMED", "required": false, "repr": "**kwargs"}], "doc": "<p>Produce message to topic asynchronously to Kafka by encoding with specified or default avro schema.</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</a></p>\n<ul>\n<li><code>topic</code> (str) : name of the topic where to produce message.</li>\n<li><code>value</code> (str|bytes): Message payload.</li>\n<li><code>key</code> (str|bytes): Message key. Default: <span class=\"name\">None</span>.</li>\n<li><code>partition</code> (int): Partition to produce to, else uses the configured built-in partitioner.</li>\n</ul>", "shortdoc": "Produce message to topic asynchronously to Kafka by encoding with specified or default avro schema.", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/producer.py", "lineno": 96}, {"name": "Purge", "args": [{"name": "group_id", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "group_id"}, {"name": "kwargs", "types": [], "defaultValue": null, "kind": "VAR_NAMED", "required": false, "repr": "**kwargs"}], "doc": "", "shortdoc": "", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/producer.py", "lineno": 127}, {"name": "Resume", "args": [{"name": "group_id", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "group_id"}, {"name": "topic_partitions", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "topic_partitions"}], "doc": "<p>Resume consumption for the provided list of partitions.</p>", "shortdoc": "Resume consumption for the provided list of partitions.", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/consumer.py", "lineno": 228}, {"name": "Seek", "args": [{"name": "group_id", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "group_id"}, {"name": "topic_partition", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "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</a></p>", "shortdoc": "https://docs.confluent.io/current/clients/confluent-kafka-python/#confluent_kafka.Consumer.seek", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/consumer.py", "lineno": 207}, {"name": "Start Consumer Threaded", "args": [{"name": "topics", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "topics"}, {"name": "group_id", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "group_id=None"}, {"name": "server", "types": [], "defaultValue": "127.0.0.1", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "server=127.0.0.1"}, {"name": "port", "types": [], "defaultValue": "9092", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "port=9092"}, {"name": "only_value", "types": [], "defaultValue": "True", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "only_value=True"}, {"name": "kwargs", "types": [], "defaultValue": null, "kind": "VAR_NAMED", "required": false, "repr": "**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</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</a> keyword.</p>\n<ul>\n<li><code>topics</code> (list): List of topics for subscription.</li>\n<li><code>group_id</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</span>.</li>\n</ul>", "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": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/consumer.py", "lineno": 307}, {"name": "Subscribe Topic", "args": [{"name": "group_id", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "group_id"}, {"name": "topics", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "topics"}, {"name": "kwargs", "types": [], "defaultValue": null, "kind": "VAR_NAMED", "required": false, "repr": "**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</a></p>\n<ul>\n<li><code>topics</code> (list): List of topics for subscription.</li>\n</ul>", "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": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/consumer.py", "lineno": 162}, {"name": "Unassign", "args": [{"name": "group_id", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "group_id"}], "doc": "", "shortdoc": "", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/consumer.py", "lineno": 193}, {"name": "Unsubscribe", "args": [{"name": "group_id", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "group_id"}], "doc": "<p>Unsubscribe of topics.</p>", "shortdoc": "Unsubscribe of topics.", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/consumer.py", "lineno": 196}], "dataTypes": {"enums": [], "typedDicts": []}}
+libdoc = {"name": "ConfluentKafkaLibrary", "doc": "<p>ConfluentKafkaLibrary is a Robot Framework library which wraps up <a href=\"https://github.com/confluentinc/confluent-kafka-python\">confluent-kafka-python</a>. Library supports more functionality like running more clients based on <span class=\"name\">group_id</span> or running them in threaded mode during the tests, decoding of gathered data etc. (<span class=\"name\">See</span> <a href=\"#Examples\" class=\"name\">Examples</a>).</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</a>. For more information about Robot Framework, see <a href=\"http://robotframework.org\">http://robotframework.org</a>.</p>\n<h3 id=\"Examples\">Examples</h3>\n<p>See <a href=\"https://github.com/robooo/robotframework-ConfluentKafkaLibrary/tree/master/examples\">repo examples</a>.</p>\n<p><b>Basic Consumer with predefined group_id</b></p>\n<table border=\"1\">\n<tr>\n<td>${group_id}=</td>\n<td><a href=\"#Create%20Consumer\" class=\"name\">Create Consumer</a></td>\n<td>group_id=mygroup</td>\n<td># if group_id is not defined uuid4() is gemerated</td>\n</tr>\n<tr>\n<td><a href=\"#Subscribe%20Topic\" class=\"name\">Subscribe Topic</a></td>\n<td>group_id=${group_id}</td>\n<td>topics=test_topic</td>\n<td></td>\n</tr>\n<tr>\n<td>${result}=</td>\n<td><a href=\"#Poll\" class=\"name\">Poll</a></td>\n<td>group_id=${group_id}</td>\n<td>max_records=5</td>\n</tr>\n<tr>\n<td><span class=\"name\">Log</span></td>\n<td>${result}</td>\n<td></td>\n<td></td>\n</tr>\n<tr>\n<td><a href=\"#Unsubscribe\" class=\"name\">Unsubscribe</a></td>\n<td>${group_id}</td>\n<td></td>\n<td></td>\n</tr>\n<tr>\n<td><a href=\"#Close%20Consumer\" class=\"name\">Close Consumer</a></td>\n<td>${group_id}</td>\n<td></td>\n<td></td>\n</tr>\n</table>\n<p><b>More Consumers</b></p>\n<table border=\"1\">\n<tr>\n<td>${group_id_1}=</td>\n<td><a href=\"#Create%20Consumer\" class=\"name\">Create Consumer</a></td>\n<td></td>\n<td></td>\n</tr>\n<tr>\n<td><a href=\"#Subscribe%20Topic\" class=\"name\">Subscribe Topic</a></td>\n<td>group_id=${group_id_1}</td>\n<td>topics=topic1</td>\n<td></td>\n</tr>\n<tr>\n<td>${group_id_2}=</td>\n<td><a href=\"#Create%20Consumer\" class=\"name\">Create Consumer</a></td>\n<td></td>\n<td></td>\n</tr>\n<tr>\n<td><a href=\"#Subscribe%20Topic\" class=\"name\">Subscribe Topic</a></td>\n<td>group_id=${group_id_2}</td>\n<td>topics=topic2</td>\n<td></td>\n</tr>\n<tr>\n<td>${result_1}=</td>\n<td><a href=\"#Poll\" class=\"name\">Poll</a></td>\n<td>group_id=${group_id_1}</td>\n<td>max_records=5</td>\n</tr>\n<tr>\n<td>${result_2}=</td>\n<td><a href=\"#Poll\" class=\"name\">Poll</a></td>\n<td>group_id=${group_id_2}</td>\n<td>max_records=2</td>\n</tr>\n<tr>\n<td><a href=\"#Unsubscribe\" class=\"name\">Unsubscribe</a></td>\n<td>${group_id_1}</td>\n<td></td>\n<td></td>\n</tr>\n<tr>\n<td><a href=\"#Unsubscribe\" class=\"name\">Unsubscribe</a></td>\n<td>${group_id_2}</td>\n<td></td>\n<td></td>\n</tr>\n<tr>\n<td><a href=\"#Close%20Consumer\" class=\"name\">Close Consumer</a></td>\n<td>${group_id_1}</td>\n<td></td>\n<td></td>\n</tr>\n<tr>\n<td><a href=\"#Close%20Consumer\" class=\"name\">Close Consumer</a></td>\n<td>${group_id_2}</td>\n<td></td>\n<td></td>\n</tr>\n</table>\n<p><b>Run Avro Consumer over HTTPS and threaded:</b></p>\n<table border=\"1\">\n<tr>\n<td>${thread}=</td>\n<td><a href=\"#Start%20Consumer%20Threaded\" class=\"name\">Start Consumer Threaded</a></td>\n<td></td>\n</tr>\n<tr>\n<td></td>\n<td>...  topics=test_topic</td>\n<td></td>\n</tr>\n<tr>\n<td></td>\n<td>...  schema_registry_url=https://localhost:8081</td>\n<td></td>\n</tr>\n<tr>\n<td></td>\n<td>...  auto_offset_reset=earliest</td>\n<td># We want to get all messages</td>\n</tr>\n<tr>\n<td></td>\n<td>...  security.protocol=ssl</td>\n<td></td>\n</tr>\n<tr>\n<td></td>\n<td>...  schema.registry.ssl.ca.location=/home/user/cert/caproxy.pem</td>\n<td></td>\n</tr>\n<tr>\n<td></td>\n<td>...  ssl.ca.location=/home/user/cert/caproxy.pem</td>\n<td></td>\n</tr>\n<tr>\n<td></td>\n<td>...  ssl.certificate.location=/home/user/cert/kafka-client-cert.pem</td>\n<td></td>\n</tr>\n<tr>\n<td></td>\n<td>...  ssl.key.location=/home/user/cert/kafka-client-key.pem</td>\n<td></td>\n</tr>\n<tr>\n<td><span class=\"name\">Log</span></td>\n<td><span class=\"name\">Execute commands which should push some data to topic</span></td>\n<td></td>\n</tr>\n<tr>\n<td>${messages}=</td>\n<td><a href=\"#Get%20Messages%20From%20Thread\" class=\"name\">Get Messages From Thread</a></td>\n<td>${thread}</td>\n</tr>\n</table>", "version": "1.7.0", "generated": "2021-11-08 10:39:22", "type": "LIBRARY", "scope": "GLOBAL", "docFormat": "HTML", "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/__init__.py", "lineno": 14, "tags": [], "inits": [], "keywords": [{"name": "Alter Configs", "args": [{"name": "group_id", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "group_id"}, {"name": "resources", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "resources"}, {"name": "kwargs", "types": [], "defaultValue": null, "kind": "VAR_NAMED", "required": false, "repr": "**kwargs"}], "doc": "<p>Update configuration properties for the specified resources.</p>\n<ul>\n<li><code>resources</code>  (list(ConfigResource) or ConfigResource): Resources to update configuration of.</li>\n</ul>", "shortdoc": "Update configuration properties for the specified resources. - ``resources``  (list(ConfigResource) or ConfigResource): Resources to update configuration of.", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/admin_client.py", "lineno": 95}, {"name": "Assign To Topic Partition", "args": [{"name": "group_id", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "group_id"}, {"name": "topic_partitions", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "topic_partitions"}], "doc": "<p>Assign a list of TopicPartitions.</p>\n<ul>\n<li><code>topic_partitions</code> (<span class=\"name\">TopicPartition</span> or list of <span class=\"name\">TopicPartition</span>): Assignment for this instance.</li>\n</ul>", "shortdoc": "Assign a list of TopicPartitions.", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/consumer.py", "lineno": 181}, {"name": "Clear Messages From Thread", "args": [{"name": "running_thread", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "running_thread"}], "doc": "<p>Remove all records gathered from specific thread</p>\n<ul>\n<li><code>running_thread</code> (Thread object) - thread which was executed with <a href=\"#Start%20Consumer%20Threaded\" class=\"name\">Start Consumer Threaded</a> keyword</li>\n</ul>", "shortdoc": "Remove all records gathered from specific thread - ``running_thread`` (Thread object) - thread which was executed with     `Start Consumer Threaded` keyword", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/consumer.py", "lineno": 346}, {"name": "Close Consumer", "args": [{"name": "group_id", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "group_id"}], "doc": "<p>Close down and terminate the Kafka Consumer.</p>", "shortdoc": "Close down and terminate the Kafka Consumer.", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/consumer.py", "lineno": 200}, {"name": "Config Resource", "args": [{"name": "restype", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "restype"}, {"name": "name", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "name"}, {"name": "kwargs", "types": [], "defaultValue": null, "kind": "VAR_NAMED", "required": false, "repr": "**kwargs"}], "doc": "<p>Represents a resource that has configuration, and (optionally) a collection of configuration properties for that resource. Used by describe_configs() and alter_configs().</p>\n<ul>\n<li><code>restype</code> (ConfigResource.Type): The resource type.</li>\n<li><code>name</code> (str): The resource name, which depends on the resource type. For RESOURCE_BROKER, the resource name is the broker id.</li>\n</ul>", "shortdoc": "Represents a resource that has configuration, and (optionally) a collection of configuration properties for that resource. Used by describe_configs() and alter_configs(). - ``restype`` (ConfigResource.Type): The resource type. -  ``name`` (str): The resource name, which depends on the resource type. For RESOURCE_BROKER,     the resource name is the broker id.", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/__init__.py", "lineno": 117}, {"name": "Create Admin Client", "args": [{"name": "group_id", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "group_id=None"}, {"name": "server", "types": [], "defaultValue": "127.0.0.1", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "server=127.0.0.1"}, {"name": "port", "types": [], "defaultValue": "9092", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "port=9092"}, {"name": "kwargs", "types": [], "defaultValue": null, "kind": "VAR_NAMED", "required": false, "repr": "**kwargs"}], "doc": "", "shortdoc": "", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/admin_client.py", "lineno": 10}, {"name": "Create Consumer", "args": [{"name": "group_id", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "group_id=None"}, {"name": "server", "types": [], "defaultValue": "127.0.0.1", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "server=127.0.0.1"}, {"name": "port", "types": [], "defaultValue": "9092", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "port=9092"}, {"name": "enable_auto_commit", "types": [], "defaultValue": "True", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "enable_auto_commit=True"}, {"name": "auto_offset_reset", "types": [], "defaultValue": "latest", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "auto_offset_reset=latest"}, {"name": "schema_registry_url", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "schema_registry_url=None"}, {"name": "auto_create_topics", "types": [], "defaultValue": "True", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "auto_create_topics=True"}, {"name": "key_deserializer", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "key_deserializer=None"}, {"name": "value_deserializer", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "value_deserializer=None"}, {"name": "legacy", "types": [], "defaultValue": "True", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "legacy=True"}, {"name": "kwargs", "types": [], "defaultValue": null, "kind": "VAR_NAMED", "required": false, "repr": "**kwargs"}], "doc": "<p>Create Kafka Consumer and returns its <span class=\"name\">group_id</span> as string.</p>\n<p>Keyword Arguments:</p>\n<ul>\n<li><code>server</code>: (str): IP address / domain, that the consumer should contact to bootstrap initial cluster metadata. Default: <span class=\"name\">127.0.0.1</span>.</li>\n<li><code>port</code> (int): Port number. Default: <span class=\"name\">9092</span>.</li>\n<li><code>group_id</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</span>.</li>\n<li><code>auto_offset_reset</code> (str): A policy for resetting offsets on OffsetOutOfRange errors: <span class=\"name\">earliest</span> will move to the oldest available message, <span class=\"name\">latest</span> will move to the most recent. Any other value will raise the exception. Default: <span class=\"name\">latest</span>.</li>\n<li><code>enable_auto_commit</code> (bool): If true the consumer's offset will be periodically committed in the background. Default: <span class=\"name\">True</span>.</li>\n<li><code>schema_registry_url</code> (str): <b>required</b> for Avro Consumer. Full URL to avro schema endpoint.</li>\n<li><code>auto_create_topics</code> (bool): Consumers no longer trigger auto creation of topics, will be removed in future release. Default: <span class=\"name\">True</span>.</li>\n<li><code>legacy</code> (bool): Activate SerializingConsumer if 'False' else AvroConsumer (legacy) is used. Will be removed when confluent-kafka will deprecate this. Default: <span class=\"name\">True</span>.</li>\n</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</a></p>", "shortdoc": "Create Kafka Consumer and returns its `group_id` as string.", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/consumer.py", "lineno": 59}, {"name": "Create Partitions", "args": [{"name": "group_id", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "group_id"}, {"name": "new_partitions", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "new_partitions"}, {"name": "kwargs", "types": [], "defaultValue": null, "kind": "VAR_NAMED", "required": false, "repr": "**kwargs"}], "doc": "<p>Create additional partitions for the given topics.</p>\n<ul>\n<li><code>new_partitions</code>  (list(NewPartitions) or NewPartitions): New partitions to be created.</li>\n</ul>", "shortdoc": "Create additional partitions for the given topics. - ``new_partitions``  (list(NewPartitions) or NewPartitions): New partitions to be created.", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/admin_client.py", "lineno": 58}, {"name": "Create Producer", "args": [{"name": "server", "types": [], "defaultValue": "127.0.0.1", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "server=127.0.0.1"}, {"name": "port", "types": [], "defaultValue": "9092", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "port=9092"}, {"name": "group_id", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "group_id=None"}, {"name": "schema_registry_url", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "schema_registry_url=None"}, {"name": "value_schema", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "value_schema=None"}, {"name": "key_schema", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "key_schema=None"}, {"name": "key_serializer", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "key_serializer=None"}, {"name": "value_serializer", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "value_serializer=None"}, {"name": "legacy", "types": [], "defaultValue": "True", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "legacy=True"}, {"name": "kwargs", "types": [], "defaultValue": null, "kind": "VAR_NAMED", "required": false, "repr": "**kwargs"}], "doc": "<p>Create Kafka Producer and returns its <span class=\"name\">group_id</span> as string. If <span class=\"name\">schema_registry_url</span> is used, Kafka Producer client which does avro schema encoding to messages is created instead.</p>\n<p>Keyword Arguments:</p>\n<ul>\n<li><code>server</code>: (str): IP address / domain, that the consumer should contact to bootstrap initial cluster metadata. Default: <span class=\"name\">127.0.0.1</span>.</li>\n<li><code>port</code> (int): Port number. Default: <span class=\"name\">9092</span>.</li>\n<li><code>schema_registry_url</code> (str): <b>required</b> for Avro Consumer. Full URL to avro schema endpoint.</li>\n<li><code>value_schema</code> (str): Optional default avro schema for value or path to file with schema. Default: <span class=\"name\">None</span></li>\n<li><code>key_schema</code> (str): Optional default avro schema for key. Default: <span class=\"name\">None</span></li>\n<li><code>legacy</code> (bool): Activate SerializingProducer if 'False' else AvroProducer (legacy) is used. Will be removed when confluent-kafka will deprecate this. Default: <span class=\"name\">True</span>.</li>\n</ul>", "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": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/producer.py", "lineno": 25}, {"name": "Create Topic Partition", "args": [{"name": "topic_name", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "topic_name"}, {"name": "partition", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "partition=None"}, {"name": "offset", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "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</a></p>\n<ul>\n<li><code>topic_name</code> (str): Topic name.</li>\n<li><code>partition</code> (int): Partition id.</li>\n<li><code>offset</code> (int): Initial partition offset.</li>\n</ul>", "shortdoc": "Returns TopicPartiton object based on https://docs.confluent.io/current/clients/confluent-kafka-python/#topicpartition", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/consumer.py", "lineno": 140}, {"name": "Create Topics", "args": [{"name": "group_id", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "group_id"}, {"name": "new_topics", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "new_topics"}, {"name": "kwargs", "types": [], "defaultValue": null, "kind": "VAR_NAMED", "required": false, "repr": "**kwargs"}], "doc": "<p>Create one or more new topics and wait for each one to finish.</p>\n<ul>\n<li><code>new_topics</code> (list(NewTopic) or NewTopic): A list of specifications (NewTopic) or a single instance for the topics that should be created.</li>\n</ul>", "shortdoc": "Create one or more new topics and wait for each one to finish. - ``new_topics`` (list(NewTopic) or NewTopic): A list of specifications (NewTopic)     or a single instance for the topics that should be created.", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/admin_client.py", "lineno": 27}, {"name": "Delete Topics", "args": [{"name": "group_id", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "group_id"}, {"name": "topics", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "topics"}, {"name": "kwargs", "types": [], "defaultValue": null, "kind": "VAR_NAMED", "required": false, "repr": "**kwargs"}], "doc": "", "shortdoc": "", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/admin_client.py", "lineno": 45}, {"name": "Describe Configs", "args": [{"name": "group_id", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "group_id"}, {"name": "resources", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "resources"}, {"name": "kwargs", "types": [], "defaultValue": null, "kind": "VAR_NAMED", "required": false, "repr": "**kwargs"}], "doc": "<p>Get the configuration of the specified resources.</p>\n<ul>\n<li><code>resources</code>  (list(ConfigResource) or ConfigResource): Resources to get the configuration for.</li>\n</ul>", "shortdoc": "Get the configuration of the specified resources. - ``resources``  (list(ConfigResource) or ConfigResource): Resources to get the configuration for.", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/admin_client.py", "lineno": 75}, {"name": "Flush", "args": [{"name": "group_id", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "group_id"}, {"name": "timeout", "types": [], "defaultValue": "0.1", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "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.</p>\n<ul>\n<li><span class=\"name\">timeout</span> (real) : Optional timeout. Default: <span class=\"name\">0.1</span>.</li>\n</ul>", "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": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/producer.py", "lineno": 109}, {"name": "Get All Consumers", "args": [], "doc": "<p>Returns all non-threaded consumers</p>", "shortdoc": "Returns all non-threaded consumers", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/consumer.py", "lineno": 135}, {"name": "Get Assignment", "args": [{"name": "group_id", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "group_id"}], "doc": "", "shortdoc": "", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/consumer.py", "lineno": 178}, {"name": "Get Avro Deserializer", "args": [{"name": "schema_str", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "schema_str"}, {"name": "schema_registry_client", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "schema_registry_client"}, {"name": "from_dict", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "from_dict=None"}], "doc": "", "shortdoc": "", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/serialization.py", "lineno": 30}, {"name": "Get Avro Serializer", "args": [{"name": "schema_str", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "schema_str"}, {"name": "schema_registry_client", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "schema_registry_client"}, {"name": "to_dict", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "to_dict=None"}, {"name": "conf", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "conf=None"}], "doc": "", "shortdoc": "", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/serialization.py", "lineno": 9}, {"name": "Get Double Deserializer", "args": [], "doc": "", "shortdoc": "", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/serialization.py", "lineno": 33}, {"name": "Get Double Serializer", "args": [], "doc": "", "shortdoc": "", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/serialization.py", "lineno": 12}, {"name": "Get Integer Deserializer", "args": [], "doc": "", "shortdoc": "", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/serialization.py", "lineno": 36}, {"name": "Get Integer Serializer", "args": [], "doc": "", "shortdoc": "", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/serialization.py", "lineno": 15}, {"name": "Get Json Deserializer", "args": [{"name": "schema_str", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "schema_str"}, {"name": "from_dict", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "from_dict=None"}], "doc": "", "shortdoc": "", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/serialization.py", "lineno": 39}, {"name": "Get Json Serializer", "args": [{"name": "schema_str", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "schema_str"}, {"name": "schema_registry_client", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "schema_registry_client"}, {"name": "to_dict", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "to_dict=None"}, {"name": "conf", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "conf=None"}], "doc": "", "shortdoc": "", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/serialization.py", "lineno": 18}, {"name": "Get Messages From Thread", "args": [{"name": "running_thread", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "running_thread"}, {"name": "decode_format", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "decode_format=None"}], "doc": "<p>Returns all records gathered from specific thread</p>\n<ul>\n<li><code>running_thread</code> (Thread object) - thread which was executed with <a href=\"#Start%20Consumer%20Threaded\" class=\"name\">Start Consumer Threaded</a> keyword</li>\n<li><code>decode_format</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</a>). Default: None.</li>\n</ul>", "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": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/consumer.py", "lineno": 333}, {"name": "Get Position", "args": [{"name": "group_id", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "group_id"}, {"name": "topic_partitions", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "topic_partitions"}], "doc": "<p>Retrieve current positions (offsets) for the list of partitions.</p>\n<ul>\n<li><code>topic_partitions</code> (<span class=\"name\">TopicPartition</span> or list of <span class=\"name\">TopicPartition</span>): Assignment for this instance.</li>\n</ul>", "shortdoc": "Retrieve current positions (offsets) for the list of partitions.", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/consumer.py", "lineno": 211}, {"name": "Get Protobuf Deserializer", "args": [{"name": "message_type", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "message_type"}], "doc": "", "shortdoc": "", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/serialization.py", "lineno": 42}, {"name": "Get Protobuf Serializer", "args": [{"name": "msg_type", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "msg_type"}, {"name": "schema_registry_client", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "schema_registry_client"}, {"name": "conf", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "conf=None"}], "doc": "", "shortdoc": "", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/serialization.py", "lineno": 21}, {"name": "Get Schema Registry Client", "args": [{"name": "conf", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "conf"}], "doc": "", "shortdoc": "", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/__init__.py", "lineno": 126}, {"name": "Get String Deserializer", "args": [{"name": "codec", "types": [], "defaultValue": "utf_8", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "codec=utf_8"}], "doc": "", "shortdoc": "", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/serialization.py", "lineno": 45}, {"name": "Get String Serializer", "args": [{"name": "codec", "types": [], "defaultValue": "utf_8", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "codec=utf_8"}], "doc": "", "shortdoc": "", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/serialization.py", "lineno": 24}, {"name": "Get Topic Partitions", "args": [{"name": "topic", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "topic"}], "doc": "<p>Returns dictionary of all TopicPartitons in topic (topic.partitions).</p>", "shortdoc": "Returns dictionary of all TopicPartitons in topic (topic.partitions).", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/consumer.py", "lineno": 156}, {"name": "Get Watermark Offsets", "args": [{"name": "group_id", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "group_id"}, {"name": "topic_partition", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "topic_partition"}, {"name": "kwargs", "types": [], "defaultValue": null, "kind": "VAR_NAMED", "required": false, "repr": "**kwargs"}], "doc": "<p>Retrieve low and high offsets for partition.</p>", "shortdoc": "Retrieve low and high offsets for partition.", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/consumer.py", "lineno": 171}, {"name": "List Topics", "args": [{"name": "group_id", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "group_id"}, {"name": "topic", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "topic=None"}], "doc": "<p>Request Metadata from cluster. Could be executed with consumer or producer group_id too.</p>\n<ul>\n<li><code>topic</code> (str):  If specified, only request info about this topic, else return for all topics in cluster.</li>\n</ul>\n<p>Default: <span class=\"name\">None</span>.</p>\n<ul>\n<li><code>group_id</code> (str): <b>required</b> id of the created consumer or producer.</li>\n</ul>", "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 return for all topics in cluster. Default: `None`. - ``group_id`` (str): *required* id of the created consumer or producer.", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/__init__.py", "lineno": 87}, {"name": "Load Schema", "args": [{"name": "data", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "data"}], "doc": "", "shortdoc": "", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/producer.py", "lineno": 15}, {"name": "New Partitions", "args": [{"name": "topic", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "topic"}, {"name": "kwargs", "types": [], "defaultValue": null, "kind": "VAR_NAMED", "required": false, "repr": "**kwargs"}], "doc": "<p>Instantiate a NewPartitions object.</p>\n<ul>\n<li><code>topic</code> (str): Topic name</li>\n</ul>", "shortdoc": "Instantiate a NewPartitions object. - ``topic`` (str): Topic name", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/__init__.py", "lineno": 111}, {"name": "New Topic", "args": [{"name": "topic", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "topic"}, {"name": "kwargs", "types": [], "defaultValue": null, "kind": "VAR_NAMED", "required": false, "repr": "**kwargs"}], "doc": "<p>Instantiate a NewTopic object. Specifies per-topic settings for passing to AdminClient.create_topics().</p>\n<ul>\n<li><code>topic</code> (str): Topic name</li>\n</ul>\n<p>Note: In a multi-cluster production scenario, it is more typical to use a replication_factor of 3 for durability.</p>", "shortdoc": "Instantiate a NewTopic object. Specifies per-topic settings for passing to AdminClient.create_topics(). - ``topic`` (str): Topic name Note: In a multi-cluster production scenario, it is more typical to use a replication_factor of 3 for durability.", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/__init__.py", "lineno": 103}, {"name": "Pause", "args": [{"name": "group_id", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "group_id"}, {"name": "topic_partitions", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "topic_partitions"}], "doc": "<p>Pause consumption for the provided list of partitions.</p>", "shortdoc": "Pause consumption for the provided list of partitions.", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/consumer.py", "lineno": 220}, {"name": "Poll", "args": [{"name": "group_id", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "group_id"}, {"name": "timeout", "types": [], "defaultValue": "1", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "timeout=1"}, {"name": "max_records", "types": [], "defaultValue": "1", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "max_records=1"}, {"name": "poll_attempts", "types": [], "defaultValue": "10", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "poll_attempts=10"}, {"name": "only_value", "types": [], "defaultValue": "True", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "only_value=True"}, {"name": "auto_create_topics", "types": [], "defaultValue": "True", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "auto_create_topics=True"}, {"name": "decode_format", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "decode_format=None"}, {"name": "fail_on_deserialization", "types": [], "defaultValue": "False", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "fail_on_deserialization=False"}], "doc": "<p>Fetch and return messages from assigned topics / partitions as list.</p>\n<ul>\n<li><code>timeout</code> (int): Seconds spent waiting in poll if data is not available in the buffer.</li>\n</ul>\n<ul>\n<li><code>max_records</code> (int): maximum number of messages to get from poll. Default: 1.</li>\n</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</span></p>\n<ul>\n<li><code>poll_attempts</code> (int): Attempts to consume messages and endless looping prevention.</li>\n</ul>\n<p>Sometimes the first messages are None or the topic could be empty. Default: <span class=\"name\">10</span>.</p>\n<ul>\n<li><code>only_value</code> (bool): Return only message.value(). Default: <span class=\"name\">True</span>.</li>\n<li><code>decode_format</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</a>). Default: None.</li>\n<li><code>auto_create_topics</code> (bool): Consumers no longer trigger auto creation of topics, will be removed in future release. If True then the error message UNKNOWN_TOPIC_OR_PART is ignored. Default: <span class=\"name\">True</span>.</li>\n<li><code>fail_on_deserialization</code> (bool): If True and message deserialization fails, will raise a SerializerError exception; on False will just stop the current poll and return the message so far. Default: <span class=\"name\">False</span>.</li>\n</ul>", "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": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/consumer.py", "lineno": 234}, {"name": "Produce", "args": [{"name": "group_id", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "group_id"}, {"name": "topic", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "topic"}, {"name": "value", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "value=None"}, {"name": "key", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "key=None"}, {"name": "kwargs", "types": [], "defaultValue": null, "kind": "VAR_NAMED", "required": false, "repr": "**kwargs"}], "doc": "<p>Produce message to topic asynchronously to Kafka by encoding with specified or default avro schema.</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</a></p>\n<ul>\n<li><code>topic</code> (str) : name of the topic where to produce message.</li>\n<li><code>value</code> (str|bytes): Message payload.</li>\n<li><code>key</code> (str|bytes): Message key. Default: <span class=\"name\">None</span>.</li>\n<li><code>partition</code> (int): Partition to produce to, else uses the configured built-in partitioner.</li>\n</ul>", "shortdoc": "Produce message to topic asynchronously to Kafka by encoding with specified or default avro schema.", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/producer.py", "lineno": 86}, {"name": "Purge", "args": [{"name": "group_id", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "group_id"}, {"name": "kwargs", "types": [], "defaultValue": null, "kind": "VAR_NAMED", "required": false, "repr": "**kwargs"}], "doc": "", "shortdoc": "", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/producer.py", "lineno": 117}, {"name": "Resume", "args": [{"name": "group_id", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "group_id"}, {"name": "topic_partitions", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "topic_partitions"}], "doc": "<p>Resume consumption for the provided list of partitions.</p>", "shortdoc": "Resume consumption for the provided list of partitions.", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/consumer.py", "lineno": 227}, {"name": "Seek", "args": [{"name": "group_id", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "group_id"}, {"name": "topic_partition", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "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</a></p>", "shortdoc": "https://docs.confluent.io/current/clients/confluent-kafka-python/#confluent_kafka.Consumer.seek", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/consumer.py", "lineno": 206}, {"name": "Start Consumer Threaded", "args": [{"name": "topics", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "topics"}, {"name": "group_id", "types": [], "defaultValue": "None", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "group_id=None"}, {"name": "server", "types": [], "defaultValue": "127.0.0.1", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "server=127.0.0.1"}, {"name": "port", "types": [], "defaultValue": "9092", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "port=9092"}, {"name": "only_value", "types": [], "defaultValue": "True", "kind": "POSITIONAL_OR_NAMED", "required": false, "repr": "only_value=True"}, {"name": "kwargs", "types": [], "defaultValue": null, "kind": "VAR_NAMED", "required": false, "repr": "**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</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</a> keyword.</p>\n<ul>\n<li><code>topics</code> (list): List of topics for subscription.</li>\n<li><code>group_id</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</span>.</li>\n</ul>", "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": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/consumer.py", "lineno": 304}, {"name": "Subscribe Topic", "args": [{"name": "group_id", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "group_id"}, {"name": "topics", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "topics"}, {"name": "kwargs", "types": [], "defaultValue": null, "kind": "VAR_NAMED", "required": false, "repr": "**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</a></p>\n<ul>\n<li><code>topics</code> (list): List of topics for subscription.</li>\n</ul>", "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": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/consumer.py", "lineno": 161}, {"name": "Unassign", "args": [{"name": "group_id", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "group_id"}], "doc": "", "shortdoc": "", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/consumer.py", "lineno": 192}, {"name": "Unsubscribe", "args": [{"name": "group_id", "types": [], "defaultValue": null, "kind": "POSITIONAL_OR_NAMED", "required": true, "repr": "group_id"}], "doc": "<p>Unsubscribe of topics.</p>", "shortdoc": "Unsubscribe of topics.", "tags": [], "source": "/home/runner/work/robotframework-ConfluentKafkaLibrary/robotframework-ConfluentKafkaLibrary/src/ConfluentKafkaLibrary/consumer.py", "lineno": 195}], "dataTypes": {"enums": [], "typedDicts": []}}
 </script>
 <title></title>
 </head>
diff --git a/src/ConfluentKafkaLibrary/__init__.py b/src/ConfluentKafkaLibrary/__init__.py
index 35f0a29..cab2a98 100644
--- a/src/ConfluentKafkaLibrary/__init__.py
+++ b/src/ConfluentKafkaLibrary/__init__.py
@@ -1,15 +1,17 @@
-from robot.libraries.BuiltIn import BuiltIn, RobotNotRunningError
 import confluent_kafka
+from confluent_kafka.schema_registry import SchemaRegistryClient
+from confluent_kafka.admin import AdminClient, NewTopic, NewPartitions, ConfigResource
+from robot.libraries.BuiltIn import BuiltIn, RobotNotRunningError
 
 from .consumer import KafkaConsumer
 from .producer import KafkaProducer
+from .admin_client import KafkaAdminClient
 from .serialization import Serializer, Deserializer
 from .version import VERSION
-from confluent_kafka.schema_registry import SchemaRegistryClient
 
 
 #class ConfluentKafkaLibrary(KafkaConsumer, KafkaProducer, Serializer, Deserializer):
-class ConfluentKafkaLibrary(KafkaConsumer, KafkaProducer, Serializer, Deserializer):
+class ConfluentKafkaLibrary(KafkaConsumer, KafkaProducer, KafkaAdminClient, Serializer, Deserializer):
     """ConfluentKafkaLibrary is a Robot Framework library which wraps up
     [https://github.com/confluentinc/confluent-kafka-python | confluent-kafka-python].
     Library supports more functionality like running more clients based on `group_id`
@@ -67,6 +69,7 @@ class ConfluentKafkaLibrary(KafkaConsumer, KafkaProducer, Serializer, Deserializ
     def __init__(self):
         KafkaConsumer.__init__(self)
         KafkaProducer.__init__(self)
+        KafkaAdminClient.__init__(self)
         self._set_globals_variables_if_robot_running()
 
     def _set_globals_variables_if_robot_running(self):
@@ -75,12 +78,15 @@ def _set_globals_variables_if_robot_running(self):
             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)
+            BuiltIn().set_global_variable('${ADMIN_RESOURCE_BROKER}', confluent_kafka.admin.RESOURCE_BROKER)
+            BuiltIn().set_global_variable('${ADMIN_RESOURCE_GROUP}', confluent_kafka.admin.RESOURCE_GROUP)
+            BuiltIn().set_global_variable('${ADMIN_RESOURCE_TOPIC}', confluent_kafka.admin.RESOURCE_TOPIC)
         except RobotNotRunningError as e:
             pass
 
     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.
+        - ``topic`` (str):  If specified, only request info about this topic, else return for all topics in cluster.
         Default: `None`.
         - ``group_id`` (str): *required* id of the created consumer or producer.
         """
@@ -94,6 +100,28 @@ def list_topics(self, group_id, topic=None):
 
         raise ValueError('Consumer or producer group_id is wrong or does not exists!')
 
+    def new_topic(self, topic, **kwargs):
+        """Instantiate a NewTopic object. Specifies per-topic settings for passing to AdminClient.create_topics().
+        - ``topic`` (str): Topic name
+        Note: In a multi-cluster production scenario, it is more typical to use a
+        replication_factor of 3 for durability.
+        """
+        return NewTopic(topic=topic, **kwargs)
+
+    def new_partitions(self, topic, **kwargs):
+        """Instantiate a NewPartitions object.
+        - ``topic`` (str): Topic name
+        """
+        return NewPartitions(topic=topic, **kwargs)
+
+    def config_resource(self, restype, name, **kwargs):
+        """Represents a resource that has configuration, and (optionally) a collection of configuration properties
+        for that resource. Used by describe_configs() and alter_configs().
+        - ``restype`` (ConfigResource.Type): The resource type.
+        -  ``name`` (str): The resource name, which depends on the resource type. For RESOURCE_BROKER,
+            the resource name is the broker id.
+        """
+        return ConfigResource(restype=restype, name=name, **kwargs)
+
     def get_schema_registry_client(self, conf):
         return SchemaRegistryClient(conf)
-
diff --git a/src/ConfluentKafkaLibrary/admin_client.py b/src/ConfluentKafkaLibrary/admin_client.py
new file mode 100644
index 0000000..a74f4af
--- /dev/null
+++ b/src/ConfluentKafkaLibrary/admin_client.py
@@ -0,0 +1,110 @@
+import uuid
+from confluent_kafka.admin import AdminClient
+
+
+class KafkaAdminClient():
+
+    def __init__(self):
+        self.admin_clients = {}
+
+    def create_admin_client(
+        self,
+        group_id=None,
+        server="127.0.0.1",
+        port="9092",
+        **kwargs
+    ):
+        if group_id is None:
+            group_id = str(uuid.uuid4())
+
+        admin_client = AdminClient({
+            'bootstrap.servers': '{}:{}'.format(server, port),
+            **kwargs})
+
+        self.admin_clients[group_id] = admin_client
+        return group_id
+
+    def create_topics(self, group_id, new_topics, **kwargs):
+        """Create one or more new topics and wait for each one to finish.
+        - ``new_topics`` (list(NewTopic) or NewTopic): A list of specifications (NewTopic)
+            or a single instance for the topics that should be created.
+        """
+        fs = None
+        if isinstance(new_topics, list):
+            fs = self.admin_clients[group_id].create_topics(new_topics, **kwargs)
+        else:
+            fs = self.admin_clients[group_id].create_topics([new_topics], **kwargs)
+
+        for topic, f in fs.items():
+            try:
+                f.result()  # The result itself is None
+                print("Topic {} created".format(topic))
+            except Exception as e:
+                raise Exception("Failed to create topic {}: {}".format(topic, e))
+
+    def delete_topics(self, group_id, topics, **kwargs):
+        if isinstance(topics, str):
+            topics = [topics]
+
+        fs = self.admin_clients[group_id].delete_topics(topics, **kwargs)
+
+        for topic, f in fs.items():
+            try:
+                f.result()  # The result itself is None
+                print("Topic {} deleted".format(topic))
+            except Exception as e:
+                raise Exception("Failed to delete topic {}: {}".format(topic, e))
+
+    def create_partitions(self, group_id, new_partitions, **kwargs):
+        """Create additional partitions for the given topics.
+        - ``new_partitions``  (list(NewPartitions) or NewPartitions): New partitions to be created.
+        """
+        fs = None
+        if isinstance(new_partitions, list):
+            fs = self.admin_clients[group_id].create_partitions(new_partitions, **kwargs)
+        else:
+            fs = self.admin_clients[group_id].create_partitions([new_partitions], **kwargs)
+
+        for topic, f in fs.items():
+            try:
+                f.result()  # The result itself is None
+                print("Additional partitions created for topic {}".format(topic))
+            except Exception as e:
+                raise Exception("Failed to add partitions to topic {}: {}".format(topic, e))
+
+    def describe_configs(self, group_id, resources, **kwargs):
+        """Get the configuration of the specified resources.
+        - ``resources``  (list(ConfigResource) or ConfigResource): Resources to get the configuration for.
+        """
+        fs = None
+        if isinstance(resources, list):
+            fs = self.admin_clients[group_id].describe_configs(resources, **kwargs)
+        else:
+            fs = self.admin_clients[group_id].describe_configs([resources], **kwargs)
+
+        for res, f in fs.items():
+            try:
+                configs = f.result()
+                return configs
+
+            except KafkaException as e:
+                raise KafkaException("Failed to describe {}: {}".format(res, e))
+            except Exception:
+                raise
+
+    def alter_configs(self, group_id, resources, **kwargs):
+        """Update configuration properties for the specified resources.
+        - ``resources``  (list(ConfigResource) or ConfigResource): Resources to update configuration of.
+        """
+        fs = None
+        if isinstance(resources, list):
+            fs = self.admin_clients[group_id].alter_configs(resources, **kwargs)
+        else:
+            fs = self.admin_clients[group_id].alter_configs([resources], **kwargs)
+
+        for res, f in fs.items():
+            try:
+                f.result()  # The result itself is None
+                print("{} configuration successfully altered".format(res))
+            except Exception:
+                raise
diff --git a/src/ConfluentKafkaLibrary/consumer.py b/src/ConfluentKafkaLibrary/consumer.py
index daa947d..1ab26ba 100644
--- a/src/ConfluentKafkaLibrary/consumer.py
+++ b/src/ConfluentKafkaLibrary/consumer.py
@@ -18,7 +18,7 @@ def __init__(
         **kwargs
     ):
 
-        super(GetMessagesThread, self).__init__()
+        super().__init__()
         self.daemon = True
         self._is_running = True
         self.only_value = only_value
@@ -102,7 +102,7 @@ def create_consumer(
         if group_id is None:
             group_id = str(uuid.uuid4())
 
-        if schema_registry_url and legacy == True:
+        if schema_registry_url and legacy:
             consumer = AvroConsumer({
                 'bootstrap.servers': '{}:{}'.format(server, port),
                 'group.id': group_id,
@@ -111,7 +111,7 @@ def create_consumer(
                 'auto.offset.reset': auto_offset_reset,
                 'schema.registry.url': schema_registry_url,
                 **kwargs})
-        elif legacy == False:
+        elif not legacy:
             consumer = DeserializingConsumer({
                 'bootstrap.servers': '{}:{}'.format(server, port),
                 'group.id': group_id,
@@ -151,8 +151,7 @@ def create_topic_partition(self, topic_name, partition=None, offset=None):
             return TopicPartition(topic_name, offset)
         elif offset is None:
             return TopicPartition(topic_name, partition)
-        else:
-            return TopicPartition(topic_name)
+        return TopicPartition(topic_name)
 
     def get_topic_partitions(self, topic):
         """Returns dictionary of all TopicPartitons in topic (topic.partitions).
@@ -257,7 +256,7 @@ def poll(
             will be removed in future release. If True then the error message UNKNOWN_TOPIC_OR_PART is ignored.
             Default: `True`.
         - ``fail_on_deserialization`` (bool): If True and message deserialization fails, will raise a SerializerError
-            exception; on False will just stop the current poll and return the message so far. Default: `False`.            
+            exception; on False will just stop the current poll and return the message so far. Default: `False`.
         """
 
         messages = []
@@ -269,9 +268,9 @@ def poll(
                 error = 'Message deserialization failed for {}: {}'.format(msg, err)
                 if fail_on_deserialization:
                     raise SerializerError(error)
-                else:
-                    print(error)
-                    break
+
+                print(error)
+                break
 
             if msg is None:
                 poll_attempts -= 1
@@ -279,10 +278,9 @@ def poll(
 
             if msg.error():
                 # Workaround due to new message return + deprecation of the "Consumers no longer trigger auto creation of topics"
-                if int(msg.error().code()) == KafkaError.UNKNOWN_TOPIC_OR_PART and auto_create_topics == True:
+                if int(msg.error().code()) == KafkaError.UNKNOWN_TOPIC_OR_PART and auto_create_topics:
                     continue
-                else:
-                    raise KafkaException(msg.error())
+                raise KafkaException(msg.error())
 
             if only_value:
                 messages.append(msg.value())
@@ -300,8 +298,7 @@ def poll(
     def _decode_data(self, data, decode_format):
         if decode_format:
             return [record.decode(str(decode_format)) for record in data]
-        else:
-            return data
+        return data
 
     # Experimental - getting messages from kafka topic every second
     def start_consumer_threaded(
diff --git a/src/ConfluentKafkaLibrary/producer.py b/src/ConfluentKafkaLibrary/producer.py
index 0fbf386..f76cbe0 100644
--- a/src/ConfluentKafkaLibrary/producer.py
+++ b/src/ConfluentKafkaLibrary/producer.py
@@ -1,27 +1,23 @@
-import sys
 import uuid
-import copy
 import os
-import json
-from confluent_kafka.avro.serializer import SerializerError
+from avro import schema
 from confluent_kafka import SerializingProducer
-from confluent_kafka import Producer, KafkaError, TopicPartition
+from confluent_kafka import Producer
 from confluent_kafka import avro
-from avro import schema
 from confluent_kafka.avro import AvroProducer
 
 
-class KafkaProducer(object):
+class KafkaProducer():
 
     def __init__(self):
         self.producers = {}
 
     def load_schema(self, data):
-        if type(data) == schema.RecordSchema:
+        if isinstance(data, schema.RecordSchema):
             data = data
         elif os.path.exists(data):
             data = avro.load(data)
-        elif type(data) is str:
+        elif isinstance(data,str):
             data = str(data)
             data = avro.loads(data)
         return data
@@ -30,7 +26,6 @@ def create_producer(
         self,
         server='127.0.0.1',
         port='9092',
-        client_id='Robot',
         group_id=None,
         schema_registry_url=None,
         value_schema=None,
@@ -49,11 +44,6 @@ def create_producer(
             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`.
         - ``schema_registry_url`` (str): *required* for Avro Consumer. Full URL to avro schema endpoint.
         - ``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`
@@ -65,7 +55,7 @@ def create_producer(
         if group_id is None:
             group_id = str(uuid.uuid4())
 
-        if schema_registry_url and legacy == True:
+        if schema_registry_url and legacy:
             if value_schema:
                 value_schema = self.load_schema(value_schema)
             if key_schema:
@@ -78,7 +68,7 @@ def create_producer(
                 default_key_schema=key_schema,
                 default_value_schema=value_schema
             )
-        elif legacy == False:
+        elif not legacy:
             producer = SerializingProducer({
                 'bootstrap.servers': '{}:{}'.format(server, port),
                 'key.serializer': key_serializer,
diff --git a/src/ConfluentKafkaLibrary/serialization.py b/src/ConfluentKafkaLibrary/serialization.py
index 081481f..7f3cdfa 100644
--- a/src/ConfluentKafkaLibrary/serialization.py
+++ b/src/ConfluentKafkaLibrary/serialization.py
@@ -1,8 +1,10 @@
 from confluent_kafka.schema_registry.avro import *
+from confluent_kafka.schema_registry.protobuf import (ProtobufSerializer, ProtobufDeserializer)
 from confluent_kafka.schema_registry.json_schema import JSONSerializer, JSONDeserializer
 from confluent_kafka.serialization import *
 
-class Serializer(object):
+
+class Serializer():
 
     def get_avro_serializer(self, schema_str, schema_registry_client, to_dict=None, conf=None):
         return AvroSerializer(schema_registry_client, schema_str, to_dict, conf)
@@ -23,7 +25,7 @@ def get_string_serializer(self, codec='utf_8'):
         return StringSerializer(codec)
 
 
-class Deserializer(object):
+class Deserializer():
 
     def get_avro_deserializer(self, schema_str, schema_registry_client, from_dict=None):
         return AvroDeserializer(schema_registry_client, schema_str, from_dict)


Test patch:
diff --git a/examples/test_adminclient.robot b/examples/test_adminclient.robot
new file mode 100644
index 0000000..7d9273d
--- /dev/null
+++ b/examples/test_adminclient.robot
@@ -0,0 +1,56 @@
+*** Settings ***
+Library  ConfluentKafkaLibrary
+Library  Collections
+
+
+*** Test Cases ***
+AdminClient Topic Creation
+    ${topic_names}=  Create List  admintesting1  admintesting2  admintesting3
+    ${topics}=  Create List
+    FOR  ${topic}  IN  @{topic_names}
+      ${topic}=  New Topic  ${topic}  num_partitions=${1}  replication_factor=${1}
+      Append To List  ${topics}  ${topic}
+    END
+
+    ${admin_client_id}=  Create Admin Client
+    Create Topics  group_id=${admin_client_id}  new_topics=${topics}
+
+    ${group_id}=  Create Consumer  auto_offset_reset=earliest
+    ${topics}=  List Topics  ${group_id}
+    FOR  ${topic}  IN  @{topic_names}
+      List Should Contain Value  ${topics}  ${topic}
+    END
+    Close Consumer  ${group_id}
+    [Teardown]  Delete Topics  ${admin_client_id}  ${topic_names}
+
+AdminClient New Partitions
+    ${topic_name}=  Set Variable  admin_testing_partition
+    ${topic}=  New Topic  ${topic_name}  num_partitions=${1}  replication_factor=${1}
+    ${admin_client_id}=  Create Admin Client
+    Create Topics  group_id=${admin_client_id}  new_topics=${topic}
+
+    ${new_parts}=  New Partitions  ${topic_name}  new_total_count=${2}
+    Create Partitions  group_id=${admin_client_id}  new_partitions=${new_parts}
+    [Teardown]  Delete Topics  ${admin_client_id}  ${topic_name}
+
+AdminClient Describe Configs
+    ${resource}=  Config Resource  ${ADMIN_RESOURCE_BROKER}  1
+    ${admin_client_id}=  Create Admin Client
+    ${config}=  Describe Configs  ${admin_client_id}  ${resource}
+
+    Should Not Be Empty  ${config}
+    ${name}=  Set Variable  ${config['offsets.commit.timeout.ms'].name}
+    ${value}=  Set Variable  ${config['offsets.commit.timeout.ms'].value}
+    Should Be Equal As Strings  ${name}  offsets.commit.timeout.ms
+    Should Be Equal As Integers  ${value}  ${5000}
+
+AdminClient Alter Configs
+    ${data}=  Create Dictionary  log.retention.ms=${54321}  # DotDict
+    ${data}=  Convert To Dictionary  ${data}                # dict
+    ${resource}=  Config Resource  ${ADMIN_RESOURCE_BROKER}  1  set_config=${data}
+    ${admin_client_id}=  Create Admin Client
+
+    Alter Configs  ${admin_client_id}  ${resource}
+    Sleep  1s
+    ${config}=  Describe Configs  ${admin_client_id}  ${resource}
+    Should Be Equal As Integers  ${54321}  ${config['log.retention.ms'].value}