Add throttle_cb support (#237) #377

rnpridgeon · 2018-05-17T15:09:52Z

No description provided.

ijuma · 2018-05-21T07:39:48Z

docs/index.rst

@@ -92,6 +92,9 @@ The Python bindings also provide some additional configuration properties:
 * ``error_cb(kafka.KafkaError)``: Callback for generic/global error events. This callback is served by
  poll().

+* ``throttle_cb({broker_name_str, broker_id_int, throttle_time_ms_int})``: Callback for thorttled request reporting.


There's a typo here.

edenhill

Overall looks good, but as discussed it should be rebased on adminapi branch.

Also:

change your code to 8 whitespace tabs
remove all unrelated whitespace diffs
and the other comments :)

edenhill · 2018-05-21T10:50:45Z

examples/integration_test.py

@@ -945,6 +1010,10 @@ def print_usage(exitcode, reason=None):
            print('=' * 30, 'Verifying Producer performance (without dr_cb)', '=' * 30)
            verify_producer_performance(with_dr_cb=False)

+        # The throttle test is utilizing the producer.
+        print('=' * 30, 'Verifying throttle_cb', '=' * 30)


Since the throttle test requires manual setup (zk, whatnot) we can't have it enabled by default in the integration tests.

If you rebase this on the adminapi you can add throttle as a test mode but remove it from the default set, you would then need to do: examples/integration_tests.py --throttle mybroker

edenhill · 2018-05-21T10:51:34Z

README.md

@@ -278,12 +278,19 @@ In order to run full test suite, simply execute:

 **Run integration tests:**

-To run the integration tests, uncomment the following line from `tox.ini` and add the paths to your Kafka and Confluent Schema Registry instances. You can also run the integration tests outside of `tox` by running this command from the source root.
+To run the confluent-kafka-python integration tests you will need access to a running Kafka cluster. In order to execute successfully client will need create, read and write topics. Additionally you will need access to Zookeeper in order to set a request quota for the client `throttled-client`.


"client throttled-client."

What is that referring to?

It's referring to the client the example sets the throttle for. I will remove that as part of moving the the example to the test README

edenhill · 2018-05-21T10:52:31Z

README.md


-    examples/integration_test.py <kafka-broker> [<test-topic>] [<schema-registry>]
+```
+kafka-configs  --zookeeper <zookeeper host >:< zookeeper port> \


I dont think we want this leave of detail in the top-level README, instead add a section about integration tests (and throttle) to tests/README.md

Remove trailing and heading spaces from <zookeeper host > and < zookeeper port>

edenhill · 2018-05-21T10:53:49Z

README.md


+Be sure to update Zookeeper, Broker and Schema Registry connection string with appropriate values. You can also run the integration tests outside of `tox` by running this command from the source root.


"source root" -> "top level directory."

edenhill · 2018-05-21T10:54:35Z

confluent_kafka/src/confluent_kafka.c

@@ -1131,6 +1131,37 @@ static void error_cb (rd_kafka_t *rk, int err, const char *reason, void *opaque)
 	CallState_resume(cs);
 }

+static void throttle_cb (rd_kafka_t *rk, const char *broker_name, int32_t broker_id,
+        int throttle_time_ms, void *opaque) {


The indentation in the existing code is mixed up, but all new (and modified) C code shall use 8 whitespaces for tabs.

edenhill · 2018-05-21T10:58:41Z

examples/integration_test.py

@@ -31,6 +31,7 @@

 try:
    from progress.bar import Bar
+


Avoid unrelated whitespace diffs.

edenhill · 2018-05-21T10:59:26Z

examples/integration_test.py

+    global throttled_requests
+    throttled_requests += 1
+
+    print('Request to broker %s[id=%d] throttled for %d ms' % (t_report.get('broker_name'),


Use proper type here and verify that object attributes are set and have the expected type.

This is handled by Py_BuildValue() in confluent_kafka.c. Now that the arguments are passed as a tuple which is immutable and accessible only though the ThrottleEvent object I'm not sure if this is still needed.

edenhill · 2018-05-21T10:59:32Z

examples/integration_test.py

@@ -386,9 +399,9 @@ def verify_producer_performance(with_dr_cb=True):
        bar.finish()

    print('# producing %d messages (%.2fMb) took %.3fs: %d msgs/s, %.2f Mb/s' %
-          (msgs_produced, bytecnt / (1024*1024), t_produce_spent,
+          (msgs_produced, bytecnt / (1024 * 1024), t_produce_spent,


Avoid unrelated whitespace diffs.

edenhill · 2018-05-21T10:59:50Z

examples/integration_test.py


    print('closing consumer')
    c.close()


+def verify_throttle_cb():
+    """ Time how long it takes to produce and delivery X messages """


Update docstring

edenhill · 2018-05-21T11:00:37Z

examples/integration_test.py

+            'client.id': 'throttled_client',
+            'throttle_cb': throttle_cb}
+
+    p = confluent_kafka.Producer(**conf)


use ..Producer(conf) instead, looks neater and we're moving away from the ** notation

…fault 'all mode'

…s for integration tests

edenhill

Good stuff!

Also needs a two unit tests:

specifying throttle_cb in the config, both positive and negative (not a callable) test.
instantiate a ThrottleEvent

edenhill · 2018-05-23T05:58:42Z

confluent_kafka/__init__.py

+
+    This class is typically not user instantiated.
+
+    :ivar broker_name:


:ivar <name> <type>: <Description>

edenhill · 2018-05-23T05:58:52Z

confluent_kafka/__init__.py

+    :ivar broker_name:
+    :ivar broker_id:
+    :ivar throttle_time_ms:
+


remove empty lines

edenhill · 2018-05-23T09:16:59Z

confluent_kafka/src/confluent_kafka.c

+		goto done;
+	}
+
+    ThrottleEvent_type = cfl_PyObject_lookup("confluent_kafka",


indent is off from here on.
Use 8 whitespaces for tabs.

edenhill · 2018-05-23T09:17:54Z

confluent_kafka/src/confluent_kafka.c

+    ThrottleEvent_type = cfl_PyObject_lookup("confluent_kafka",
+                                              "ThrottleEvent");
+
+    if(!ThrottleEvent_type) {


space before paren if (

edenhill · 2018-05-23T09:18:47Z

confluent_kafka/src/confluent_kafka.c

+        goto err;
+    }
+
+    args = Py_BuildValue("(sii)", broker_name, broker_id, throttle_time_ms);


I think it is customary to express time as float seconds in Python, thus a float formatter and 1000.0 * throttle_time_ms

edenhill · 2018-05-23T09:29:49Z

tests/README.md

+
+    tox -- --<test mode>
+
+The throttle_cb integration test require an additiona step and as such are not included in the default test mode.


"default test modes"

edenhill · 2018-05-23T09:30:34Z

tests/README.md

+    kafka-configs  --zookeeper <zookeeper host>:<zookeeper port> \
+        --alter --add-config 'request_percentage=01' \
+        --entity-name throttled_client --entity-type clients
+


out of curio, is it possible to do this with alterConfigs now?

possibly, worth a try I suppose. I'll push the other review changes first then revisit

Not until 2.0 it would seem

https://cwiki.apache.org/confluence/display/KAFKA/KIP-257+-+Configurable+Quota+Management

edenhill · 2018-05-23T09:30:50Z

tests/README.md

+
+Once the throttle has been set you can proceed with the following command:
+
+    tox -- --throttle


Also include how to remove the throttle setting from zk

edenhill · 2018-05-23T09:31:04Z

tox.ini

@@ -10,7 +10,8 @@ setenv =
 commands =
    pip install -v .
    py.test -v --timeout 20 --ignore=tmp-build --import-mode append {posargs}
-    #python examples/integration_test.py [yourhost:yourport] confluent-kafka-testing http://yourhost:yourport
+    # See tests/README.md for addition notes on testing


edenhill · 2018-05-23T09:32:46Z

confluent_kafka/__init__.py

+
+        self.broker_name = broker_name
+        self.broker_id = broker_id
+        self.throttle_time_ms = throttle_time_ms


Add a str implementation

rnpridgeon · 2018-05-23T11:32:11Z

additional unit tests to follow

rnpridgeon · 2018-05-23T18:58:02Z

@edenhill I believe this completes the laundry list, inviting yet another

edenhill

Uses a mix of tabs and whitespaces for indention, please use 8 whitespaces only

edenhill · 2018-05-24T17:39:25Z

confluent_kafka/__init__.py

+class ThrottleEvent (object):
+    """
+    ThrottleEvent passed as an argument to throttle_cb.
+    Contains details about requests throttled by the broker.


It contains information pertaining to one single throttled request, don't know how to formulate that properly.

edenhill · 2018-05-24T17:40:17Z

confluent_kafka/__init__.py

+    This class is typically not user instantiated.
+
+    :ivar broker_name: The hostname of the broker which throttled the request
+    :ivar broker_id: The broker's id


I think "The broker id" is used more often

edenhill · 2018-05-24T17:40:50Z

confluent_kafka/__init__.py

+
+    :ivar broker_name: The hostname of the broker which throttled the request
+    :ivar broker_id: The broker's id
+    :ivar throttle_time_ms: The delay associated with the request(seconds)


"The amount of time the broker throttled (delayed) the request before responding (seconds)"

edenhill · 2018-05-24T17:41:44Z

confluent_kafka/__init__.py

+    """
+    def __init__(self, broker_name=None,
+                 broker_id=-1,
+                 throttle_time_ms=-1.):


haven't seen that notation before, why not -1.0?

Also, do we really need the default values? I think not

edenhill · 2018-05-24T17:43:43Z

confluent_kafka/__init__.py

+        self.throttle_time_ms = throttle_time_ms
+
+    def __str__(self):
+        return "{}:{}/{}".format(self.broker_name, self.broker_id, self.throttle_time_ms)


"host:port" is a common notation which this conflicts with, also
the broker identity is formatted as "host:port/id" in librdkafka (which is what the user will see in log messages), and it isn't really clear what the throttle time is here.
I suggest being more explicit:
"broker/id throttled for X ms"

edenhill · 2018-05-24T18:00:31Z

tests/README.md

+
+    tox -- --producer
+
+The throttle_cb integration test require an additional step and as such are not included in the default test modes.


'is not included'? singularis test

edenhill · 2018-05-24T18:01:01Z

tests/README.md

+        --alter --add-config 'request_percentage=01' \
+        --entity-name throttled_client --entity-type clients
+
+To remove the throttle you can execute the following.


edenhill · 2018-05-24T18:02:10Z

tests/test_misc.py

+
+    try:
+        confluent_kafka.Producer(conf)
+    except(TypeError) as e:


except TypeError as e

But you should use:

pytest.raises(TypeError): confluent_kafka.Producer({'throttle_cb': 1})

to have pytest handle this for you

edenhill · 2018-05-24T18:02:41Z

tests/test_misc.py

+    """ Ensure we can configure a proper callback"""
+
+    def throttle_cb(throttle_event):
+        """NOOP"""


this should be a pass

edenhill · 2018-05-24T18:03:23Z

tests/test_misc.py

+
+def test_throttleevent_instantiate():
+    """ Instantiate a ThrottleEvent """
+    assert isinstance(confluent_kafka.ThrottleEvent(), confluent_kafka.ThrottleEvent)


I'd rather see you using correct parameters, and then verifying those parameters (type and value) and verifying the str:er

rnpridgeon · 2018-05-25T23:10:47Z

19th time is the charm I always say

edenhill

Almost there!

edenhill · 2018-05-29T06:50:56Z

confluent_kafka/__init__.py

+
+    This class is typically not user instantiated.
+
+    :ivar broker_name: The hostname of the broker which throttled the request


add types as so:
:ivar broker_name str: The hostname..

edenhill · 2018-05-29T06:54:50Z

confluent_kafka/src/confluent_kafka.c

+                goto done;
+        }
+
+ /**


edenhill · 2018-05-29T06:55:52Z

docs/index.rst

+  ``client.poll()`` or ``producer.flush()``.
+
+* ``throttle_cb(confluent_kafka.ThrottleEvent)``: Callback for throttled request reporting.
+  This callback is served upon calling ``client.poll()`` or ``producer.flush()``.

 * ``stats_cb(json_str)``: Callback for statistics data. This callback is triggered by poll()


"by poll() or flush()"

edenhill · 2018-05-29T06:56:12Z

docs/index.rst


 * ``stats_cb(json_str)``: Callback for statistics data. This callback is triggered by poll()
  every ``statistics.interval.ms`` (needs to be configured separately).
  Function argument ``json_str`` is a str instance of a JSON document containing statistics data.
+  This callback is served upon calling ``client.poll()`` or ``producer.flush()``.


Add:
"See https://github.com/edenhill/librdkafka/wiki/Statistics"

edenhill · 2018-05-29T06:57:24Z

examples/integration_test.py

+    # validate argument type
+    assert isinstance(throttle_event.broker_name, str)
+    assert isinstance(throttle_event.broker_id, int)
+    assert isinstance(throttle_event.broker_name, float)


Must be a a very buoyant broker name that passes that assert

edenhill · 2018-05-29T06:57:57Z

examples/integration_test.py

+def verify_throttle_cb():
+    """ Verify throttle_cb is invoked
+        This test requires client quotas be configured.
+        See tests/README for more information


tests/README.md

just testing you

edenhill · 2018-05-29T06:59:00Z

tests/README.md

@@ -1,8 +1,51 @@
 Unit tests
 ==========

+**WARNING**: These tests require an active Kafka cluster and will create new topics.


The unit tests wont, the integration tests will, move this.

edenhill · 2018-05-29T07:00:13Z

tests/README.md

+        --alter --add-config 'request_percentage=01' \
+        --entity-name throttled_client --entity-type clients
+
+To remove the throttle you can execute the following


Move this below the tox test

edenhill · 2018-05-29T07:01:03Z

tests/test_misc.py

+    confluent_kafka.Producer({'throttle_cb': throttle_cb})
+
+
+def test_throttle_event_types():


edenhill · 2018-05-29T19:42:35Z

tests/README.md

+    tox -- --throttle
+
+
+    To remove the throttle you can execute the following


This line should not be indented (turns it into monospace).
Also remove extra newline after this line

edenhill · 2018-05-29T19:53:20Z

README.md


+    See [tests/README.md](tests/README.md) for instructions on how to run integration tests.


Look at the rendered file and you'll see that 4 space indent gives you monospace, which is not intended here. 0 indent! :)

edenhill · 2018-05-29T19:55:53Z

confluent_kafka/__init__.py

+
+    :ivar broker_name str: The hostname of the broker which throttled the request
+    :ivar broker_id int: The broker id
+    :ivar throttle_time float: The amount of time the broker throttled (delayed) the request(seconds)


I would probably do: "The amount of time (in seconds) the ..", or at least add a space before the paren: "request (seconds)"

edenhill · 2018-05-29T19:59:19Z

examples/integration_test.py

+def verify_throttle_cb():
+    """ Verify throttle_cb is invoked
+        This test requires client quotas be configured.
+        See tests/README for more information


edenhill · 2018-05-29T20:00:08Z

testit.sh

@@ -0,0 +1,15 @@
+KAFKA_HOME=$HOME/CPKAFKA


did you really mean to commit this file?

edenhill · 2018-05-29T20:00:47Z

tests/test_misc.py

+    confluent_kafka.Producer({'throttle_cb': throttle_cb})
+
+
+def test_throttle_event_types():


edenhill

LGTM!

edenhill · 2018-05-30T18:25:21Z

tests/README.md

@@ -15,19 +15,19 @@ Integration tests

 To run all of the integration test `modes` uncomment the following line from `tox.ini` and add the addresses to your Kafka and Confluent Schema Registry instances.

-    #python examples/integration_test.py {posargs} <bootstrap-servers> confluent-kafka-testing [<schema-registry-url>]
+    #python examples/integration_test.py <bootstrap-servers> confluent-kafka-testing [<schema-registry-url>]

 You can also run the integration tests outside of `tox` by running this command from the source root directory

    examples/integration_test.py <kafka-broker> [<test-topic>] [<schema-registry>]

 To run individual integration test `modes` with `tox` use the following syntax


"with tox" is no longer correct

edenhill · 2018-06-01T07:04:13Z

tests/test_misc.py

+
+def throttle_cb_instantiate_fail():
+    """ Ensure noncallables raise TypeError"""
+    with pytest.raises(TypeError):


I was told to change input type errors to ValueErrors

edenhill · 2018-06-01T07:06:36Z

Suggest squashing this (on merge, or manually) into one commit.

rnpridgeon requested a review from edenhill May 17, 2018 15:09

ijuma reviewed May 21, 2018

View reviewed changes

edenhill suggested changes May 21, 2018

View reviewed changes

rnpridgeon changed the base branch from master to adminapi May 22, 2018 18:53

rnpridgeon added 4 commits May 22, 2018 19:31

Add throttle_cb support (#237)

2186175

Refactor throttle_cb to take ThorttleEvent; seperate throttle from de…

a60230b

…fault 'all mode'

minor tweaks

abd2139

Fix-up testing documentation and minor tweak tox.ini to accept posarg…

909483c

…s for integration tests

edenhill suggested changes May 23, 2018

View reviewed changes

address review comments

1a29aa6

add unit tests, a bit more clean up

63cb2eb

edenhill suggested changes May 24, 2018

View reviewed changes

rnpridgeon added 4 commits May 25, 2018 14:06

address review concerns

f31772f

fix comment style

e3cffbf

missing flush()

9b1940c

fix print

69a5c68

edenhill suggested changes May 29, 2018

View reviewed changes

rnpridgeon added 3 commits May 29, 2018 13:56

more review changes

44178c4

comment spacing

027d7a2

fix flake failure

74e437f

edenhill suggested changes May 29, 2018

View reviewed changes

rnpridgeon added 3 commits May 29, 2018 16:10

update readme path

d3fcdf3

more review

7e7ffb8

posargs don't play nice between test types.

ba32fb7

edenhill approved these changes Jun 1, 2018

View reviewed changes

fix docs, update error type

c4193e4

rnpridgeon merged commit d52858c into confluentinc:adminapi Jun 2, 2018

rnpridgeon mentioned this pull request Jun 7, 2018

Does this library support the config option throttle_cb? #237

Closed


		Be sure to update Zookeeper, Broker and Schema Registry connection string with appropriate values. You can also run the integration tests outside of `tox` by running this command from the source root.


		This class is typically not user instantiated.

		:ivar broker_name:


		tox -- --<test mode>

		The throttle_cb integration test require an additiona step and as such are not included in the default test mode.


		Once the throttle has been set you can proceed with the following command:

		tox -- --throttle


		tox -- --producer

		The throttle_cb integration test require an additional step and as such are not included in the default test modes.


		This class is typically not user instantiated.

		:ivar broker_name: The hostname of the broker which throttled the request

		confluent_kafka.Producer({'throttle_cb': throttle_cb})


		def test_throttle_event_types():

		tox -- --throttle


		To remove the throttle you can execute the following


		See [tests/README.md](tests/README.md) for instructions on how to run integration tests.

Add throttle_cb support (#237) #377

Add throttle_cb support (#237) #377

Conversation

rnpridgeon commented May 17, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

edenhill left a 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

edenhill left a 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

rnpridgeon commented May 23, 2018

rnpridgeon commented May 23, 2018

edenhill left a 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

rnpridgeon commented May 25, 2018

edenhill left a 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