Topic Admin API, list_topics(), Python 3.7, fixes #382

edenhill · 2018-05-21T15:08:22Z

For an example what the admin api looks like in practice, see examples/adminapi.py

* Add 'get_metadata' to Producer and Consumer. * Expose metadata in a way similar to the java API

Added tests for list_topics() This also removes the rich broker objects at each reference point (such as leader, isrs, etc), since it is possible that the broker does not exist. Instead we ask the user to check that the broker id exists in ClusterMetadata.brokers before referencing.

.. and a bool refactor.

…TF8) .. and a bunch of header test fixes.

mhowlett · 2018-05-21T22:01:44Z

.travis.yml

  - os: linux
    language: python
    dist: trusty
    python: "2.7"
-    env: LD_LIBRARY_PATH="$PWD/tmp-build/lib" LIBRDKAFKA_VERSION=v0.9.5
-  # Source package verification with Python 3.5 and librdkafka master
+    env: LD_LIBRARY_PATH="$PWD/tmp-build/lib" LIBRDKAFKA_VERSION=adminapi


assume you didn't mean to check these changes in?

mhowlett · 2018-05-22T00:50:19Z

confluent_kafka/__init__.py

+        except Exception as e:
+            # Request-level exception, raise the same for all topics
+            for topic, fut in futmap.items():
+                fut.set_exception(e)


👍if topic-level results could potentially complete at different times, i think what you're doing here is definitely the right (only?) way to report request level errors and since your api is written for that possibility, this seems good for me.

something i'm wondering about elsewhere is whether or not this is the best approach for an API that only has a request level future (and thinking yes, even in that scenario this is an optimal compromise).

mhowlett · 2018-05-22T00:53:25Z

examples/adminapi.py

+import sys
+
+
+def example_create_topics(a, topics):


the examples are all synchronous - it would be good to see how this works where you're not blocking on the results to the admin call (for my own Python education as much as anything).

Made a contrived delta_alter_configs example

rnpridgeon · 2018-05-21T15:14:49Z

.travis.yml

  - os: linux
    language: python
    dist: trusty
    python: "2.7"
-    env: LD_LIBRARY_PATH="$PWD/tmp-build/lib" LIBRDKAFKA_VERSION=master
-  # Source package verification with Python 2.7 and librdkafka v0.9.5
+    env: LD_LIBRARY_PATH="$PWD/tmp-build/lib" LIBRDKAFKA_VERSION=adminapi


I don't know enough about Travis CI to say definitively but these changes seem like they should stay on the adminapi branch and not be merged to master.

Yes, but before the librdkafka adminapi branch is merged into master this is what's needed.

rnpridgeon · 2018-05-21T15:16:41Z

confluent_kafka/__init__.py


+import concurrent.futures


At 600 or so LOC does it not make sense to split this into its own file within the confluent-kafka directory? i.e. admin.py

Maybe, another question if we want to the admin sub-namespace as in the Java API.
confluent_kafka.admin.AdminClient

I'm also okay with this, cramming it all into init.py just seems a bit messy

a sub package seems better here to me, and also consistent with other bindings.

so confluent_kafka.admin.?

rnpridgeon · 2018-05-21T17:39:24Z

examples/adminapi.py

+            raise
+
+
+def example_alter_configs(a, args):


Does it make sense to include an example using the ConfigResource returned from describe_configs given the limitations imposed by Broker versions < 2.0?

rnpridgeon · 2018-05-21T17:56:40Z

confluent_kafka/__init__.py

+    :ivar orig_broker_id: The broker this metadata originated from.
+    :ivar orig_broker_name: Broker name/address this metadata originated from.
+
+    This class is typically not user instantiated.


This is inconsistent with ConfigEntry which places the disclaimer above the instance variables. I personally feel placing the disclaimer first is the right move.

https://github.com/confluentinc/confluent-kafka-python/blob/adminapi/confluent_kafka/__init__.py#L43

Note: ConfigEntry also includes this message below the constructor declaration as well.

https://github.com/confluentinc/confluent-kafka-python/blob/adminapi/confluent_kafka/__init__.py#L86

rnpridgeon · 2018-05-21T17:59:55Z

confluent_kafka/__init__.py

+    :ivar host: Broker hostname.
+    :ivar port: Broker port.
+
+    This class is typically not user instantiated.


Same as above, we should be consistent about where we document this

rnpridgeon · 2018-05-21T18:34:43Z

confluent_kafka/__init__.py

+    :ivar partitions: Map of partitions indexed by partition id. Value is PartitionMetadata object.
+    :ivar error: Topic error, or None. Value is a KafkaError object.
+
+    This class is typically not user instantiated.


See comment above about disclaimer placement

rnpridgeon · 2018-05-21T19:26:55Z

confluent_kafka/__init__.py

+              in ClusterMetadata.brokers. Always check the availability
+              of a broker id in the brokers dict.
+
+    This class is typically not user instantiated.


same as above

mhowlett · 2018-05-22T17:24:56Z

confluent_kafka/__init__.py

+        return "%s=\"%s\"" % (self.name, self.value)
+
+    @classmethod
+    def config_source_to_str(cls, source):


this seems a bit clunky - can you use an IntEnum instead?

Not available in Py2.7

the 3.4 implementation has been backported and i'm lead to believe commonly used in 2.7 libraries. can this not be leveraged?

https://pypi.org/project/enum34/

mhowlett · 2018-05-22T18:37:30Z

examples/adminapi.py

+    fs = a.describe_configs(resources)
+
+    # Wait for operation to finish.
+    for res, f in fs.items():


this for loop is interesting because it's how you'd typically want to handle these results - iterate through them. in Java (and C# traditionally) it's easier/nicer to do this if the return type is a list rather than a map, and for that reason I think a list would be better here in those languages even though a map models the data structure more correctly. but anyway, that doesn't apply in python: a map is perfect.

rnpridgeon · 2018-05-30T13:40:06Z

confluent_kafka/__init__.py

+        and any configuration parameter not specified will be reverted to
+        its default value.
+
+        With alter_configs(incremental=True) (requires broker version >=2.0),


Use the KIP or Jira number as opposed to the version number. I Don't think we can say with absolute certainty that this will make the 2.0 release just yet

rnpridgeon · 2018-05-30T13:42:14Z

confluent_kafka/__init__.py

+                  without altering the configuration. Default: False
+        :param incremental bool: If true, only update the specified configuration
+                  entries, not reverting unspecified configuration.
+                  This requires broker version >=2.0. Default: False


Use KIP and/or Jira reference

rnpridgeon · 2018-05-30T13:42:58Z

confluent_kafka/__init__.py

+                  the provided resources with the new configuration given,
+                  reverting all other configuration for the resource back
+                  to their default values.
+                  Use incremental=True to change the behaviour so that only the


Use KIP and/or Jira reference

rnpridgeon · 2018-05-30T13:45:03Z

confluent_kafka/src/Admin.c

+        float request_timeout;    /* parser: f */
+        float operation_timeout;  /* parser: f */
+        int   broker;             /* parser: i */
+        int   incremental;        /* needs special bool parsing */


worth noting version constraints here as well since its specifically mentioned in the doc string below

This code doesn't know if incremental is supported or not but relies on librdkafka to reject the option if not.

rnpridgeon · 2018-05-30T13:47:57Z

confluent_kafka/src/Admin.c

+
+        c_options = rd_kafka_AdminOptions_new(self->rk, forApi);
+        if (!c_options) {
+                PyErr_Format(PyExc_RuntimeError,


I believe illegal arguments should raise PyExc_ValueError

https://docs.python.org/2/library/exceptions.html#exceptions.ValueError

There is no user-supplied parameter here so ValueError wouldn't make sense.
This exception is only raised if the python client version is newer than librdkafka, which shouldnt happen.

rnpridgeon · 2018-05-30T13:53:34Z

confluent_kafka/src/Admin.c

+
+                if (!PyList_Check(replicas) ||
+                    (replica_cnt = (size_t)PyList_Size(replicas)) < 1) {
+                        PyErr_Format(


Likewise I'm fairly certain this should raise ValueError. TypeError exceptions as I understand it are for performing illegal operations against some type.

https://docs.python.org/2/library/exceptions.html#exceptions.TypeError

You could just as well argue that by passing an invalid type we are performing illegal operations against the argument though I suppose. Food for thought I suppose

rnpridgeon · 2018-05-30T14:13:05Z

confluent_kafka/src/Admin.c

+                return NULL;
+
+        if (!PyList_Check(topics) || (tcnt = (int)PyList_Size(topics)) < 1) {
+                PyErr_SetString(PyExc_TypeError,


ValueError?

rnpridgeon · 2018-05-30T14:26:51Z

confluent_kafka/src/Admin.c

+        /* Increase refcount for the return value since
+         * it is currently a borrowed reference. */
+        Py_INCREF(future);
+        return future;


Is it necessary to return the future here? The calling party should already have a reference to this future which is passed into the function to begin with. As long as the reference is incremented the future should still be safe once the original reference falls out of scope.

This was from an earlier incarnation where the future was set up in C.
Changed to return None

rnpridgeon · 2018-05-30T14:38:38Z

confluent_kafka/src/Admin.c

+                            &options.validate_only))
+                return NULL;
+
+        if (incremental_obj &&


Is there a good way to protect against setting incremental before its supported by Kafka?

librdkafka enforces it

…example

mhowlett · 2018-05-31T21:52:33Z

confluent_kafka/__init__.py

@@ -5,6 +5,9 @@
                    Message,
                    Producer,
                    TopicPartition,
+                    AdminClientImpl,


given you don't really want AdminClientImpl exposed, I think you probably need to provide a different definition of __all__.

I'm submoduling AdminClient into .admin.

I think AdminClientImpl still needs to implicitly exported since AdminClient is a subclass

there's the mistake there - composition is much more appropriate here I think.

Not sure what composition buys us in this case.
The only reason there is a Python AdminClient is because it makes it easier to wrap the single-future C AdminClientImpl API into a multiple feature Python API.

I'm confused about __all__ above. Instead I think i meant to suggest calling AdminClientImpl it _AdminClientImpl.

mhowlett · 2018-05-31T22:27:45Z

confluent_kafka/__init__.py

+                         RESOURCE_BROKER: "broker"}
+    _res_type_by_name = {v: k for k, v in _res_name_by_type.items()}
+
+    def __init__(self, restype, name,


an enum seems better to me for restype too, based on my research just now but no actual prior experience guiding me in what's normal. given it's an input rather than output thing though, not as important as CONFIG_SOURCE (but they should be consistent).

mhowlett · 2018-06-01T23:08:09Z

confluent_kafka/__init__.py

+        f, futmap = AdminClient._make_futures(topics, None,
+                                              AdminClient._make_topics_result)
+
+        super(AdminClient, self).delete_topics(topics, f, **kwargs)


overriding methods like this seems confusing, better to use composition.

I'm not an OO guy, as you are very well aware of. So please educate me what the practical difference is between composition and inheritance in this case?

mhowlett · 2018-06-04T15:03:15Z

confluent_kafka/admin/__init__.py

+Kafka Admin client: create, view, alter, delete topics and resources.
+"""
+from ..cimpl import (KafkaException, # noqa
+                     AdminClientImpl,


is it better to call this _AdminClientImpl?

mhowlett · 2018-06-04T16:05:45Z

confluent_kafka/admin/__init__.py

+
+        return f, futmap
+
+    def create_topics(self, new_topics, **kwargs):


The main thing I was noting about this before is that the signatures of the methods create_topics etc. change in the subclass. The new methods conveniently hide the old ones because of **kwargs, but it seems like kind of a trick (and my python's not strong enough to know if there are any downsides) and i'm a bit averse to tricks. Generally, the main disadvantage of inheritance is that coupling is stronger - everything exposed in the base classes flow through to subclasses. Except for the hiding noted above, this would apply here. The other thing that threw me was use of the suffix Impl in the super class because it implies to me more of a complete implementation, and I think a bit unusual in this context.

That said, I agree an inheritance relationship would be ok here. My complaint about signature differences + exposing of adminclientimpl details would be mitigated by use of Pythons poor excuse for protected methods - maybe change AdminClientImpl's create_topics etc. to _create_topics_impl etc. Though, the public interface is worse in that they are visible outside the class (though by convention ignored). Also maybe change the Impl suffix to Base. I don't particularly like that either, but it's more conventional.

Alternatively, I still think composition is better here. I'd be somewhat better externally. There'd just be one externally visible property _impl of type _AdminClientImpl. And I like the name _AdminClientImpl in this context.

we're no longer backwards compatible with older librdkafka

tbsaunde and others added 16 commits May 18, 2018 13:53

Add list_topics() API (#161, @tbsaunde, @stephan-hof)

b7439a7

* Add 'get_metadata' to Producer and Consumer. * Expose metadata in a way similar to the java API

Python 3.7 support

f6cda94

Topic Admin API

bb0f00c

travis: use adminapi librdkafka branch

bb39363

Require futures on Python 2.x

0a67983

Added (optional) memory tracking to integration tests

3d1e0a6

Fixes and refactoring for integration tests

ac3f603

test_docs: skip imported modules and inherited objects

4bd840c

Admin API: return multiple futures, bugfixes, docs

dad896d

Added Py2&Py3 compatible cfl_PyInt abstraction

64d7fa9

Fix memory leaks in generic setters

4c6de0d

.. and a bool refactor.

Fix call ordering to avoid callback crash on implicit close (#265)

f669580

Producer: Header values may now be unicode strings (auto convert to U…

f821652

…TF8) .. and a bunch of header test fixes.

Fix KafkaError refcounting

136efd3

Python 2/3 compat, minor test improvements

3b4f3fb

edenhill added bug enhancement labels May 21, 2018

edenhill requested review from mhowlett and rnpridgeon May 21, 2018 15:08

mhowlett reviewed May 21, 2018

View reviewed changes

mhowlett reviewed May 22, 2018

View reviewed changes

rnpridgeon suggested changes May 22, 2018

View reviewed changes

mhowlett reviewed May 22, 2018

View reviewed changes

edenhill added 2 commits May 25, 2018 15:14

Updated for librdkafka's internal -> background queue rename

bd72bf6

Require librdkafka >=v0.11.5

a08db1c

rnpridgeon reviewed May 30, 2018

View reviewed changes

Change TypeError to ValueError for invalid input

c87cc7f

edenhill added 5 commits May 31, 2018 16:42

Add __len__ to ConfigResource and overwrite option to .set_config

4f0f69e

librdkafka AdminOptions_new interface update

868787d

Don't return the future from the C Admin methods, not needed

3aa92b3

adminapi example: added an async and delta/incremental alter_configs …

41bb5a7

…example

Python 2 safe unicode in source

b01e50b

mhowlett reviewed May 31, 2018

View reviewed changes

mhowlett reviewed Jun 1, 2018

View reviewed changes

Ryan P and others added 4 commits June 1, 2018 23:00

Add throttle_cb support (#237) (#377)

d52858c

Move Admin API to admin sub module, use enums, fix doc strings

e4eeaa1

Include avro in docs, fix docstrings in confluent_kafka, Consumer, avro

d613894

Added ThrottleEvent to docs

1266b3c

mhowlett reviewed Jun 4, 2018

View reviewed changes

edenhill added 3 commits June 5, 2018 10:41

Rename AdminClientImpl -> _AdminClientImpl

2f32034

travis: use librdkafka master branch instead of adminapi

6d928d2

travis: remove librdkafka v0.11.4

8d1e8d3

we're no longer backwards compatible with older librdkafka

edenhill merged commit 03136d5 into master Jun 7, 2018

rnpridgeon mentioned this pull request Jun 18, 2018

Python 3 - undefined symbol: PyString_AsStringAndSize #383

Closed

7 tasks

rnpridgeon mentioned this pull request Jun 29, 2018

KIP-4 - CreateTopics API #244

Closed


		return f, futmap

		def create_topics(self, new_topics, **kwargs):

Topic Admin API, list_topics(), Python 3.7, fixes #382

Topic Admin API, list_topics(), Python 3.7, fixes #382

Conversation

edenhill commented May 21, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mhowlett May 22, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mhowlett May 22, 2018 •

edited

Loading