Skip to content

kafkatest, refactoring, and bug fixes #1

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
edenhill merged 13 commits into from
May 10, 2016
Merged

kafkatest, refactoring, and bug fixes #1

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
77305ff
Added (global) on_commit callback.
edenhill Apr 25, 2016
5433bb8
Raise exception on rebalance_cb arg building failure
edenhill Apr 25, 2016
4592b31
Improved error formatting
edenhill Apr 25, 2016
6d90fa7
Renamed producer delivery_callback to on_delivery
edenhill Apr 29, 2016
e9c9570
Minor formatting cleanup
edenhill Apr 29, 2016
b520c8d
Adjust consumer test error codes and timeouts
edenhill Apr 29, 2016
b2568b4
Restructured: C bindings are now a submodule (cimpl)
edenhill Apr 29, 2016
0dad9ef
Added kafkatest verifiable consumer and producer
edenhill Apr 29, 2016
99b774b
on_commit(): removed consumer arg
edenhill May 9, 2016
e6d3877
Minor modifications following Ewen's code review
edenhill May 9, 2016
6936d9f
Moved sources to src/ to avoid cimpl package collision
edenhill May 9, 2016
ee588bf
Minor cleanups
edenhill May 9, 2016
f853b46
test_docs: dont require module docs (cimpl)
edenhill May 9, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
on_commit(): removed consumer arg
  • Loading branch information
@edenhill
edenhill committed May 9, 2016
commit 99b774b9bcecaad009dc5c492e00195443d23166
4 changes: 2 additions & 2 deletions confluent_kafka/cimpl/Consumer.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -552,7 +552,7 @@ PyTypeObject ConsumerType = {
"request has succeeded or failed.\n"
"\n"
"\n"
".. py:function:: on_commit(consumer, err, partitions)\n"
".. py:function:: on_commit(err, partitions)\n"
"\n"
" :param Consumer consumer: Consumer instance.\n"
Copy link
Contributor Author

@ewencp ewencp Apr 30, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I can't remember if passing the consumer instance as part of the callback came up before. I checked and we have it elsewhere. Is this valuable? Often in Python you just grab what you need via closure.

" :param KafkaError err: Commit error object, or None on success.\n"
Expand Down Expand Up @@ -656,7 +656,7 @@ static void Consumer_offset_commit_cb (rd_kafka_t *rk, rd_kafka_resp_err_t err,
/* Construct list of TopicPartition based on 'c_parts' */
parts = c_parts_to_py(c_parts);

args = Py_BuildValue("(OOO)", self, k_err, parts);
args = Py_BuildValue("(OO)", k_err, parts);

Py_DECREF(k_err);
Py_DECREF(parts);
Expand Down
2 changes: 1 addition & 1 deletion confluent_kafka/kafkatest/kafkatest_verifiable_consumer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -105,7 +105,7 @@ def on_revoke (self, consumer, partitions):
self.do_commit(immediate=True)


def on_commit (self, consumer, err, partitions):
def on_commit (self, err, partitions):
""" Offsets Committed callback """
if err is not None and err.code() == KafkaError._NO_OFFSET:
self.dbg('on_commit(): no offsets to commit')
Expand Down
2 changes: 1 addition & 1 deletion examples/integration_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -191,7 +191,7 @@ def verify_producer_performance(with_dr_cb=True):
(t_delivery_spent - t_produce_spent))


def print_commit_result (consumer, err, partitions):
def print_commit_result (err, partitions):
if err is not None:
print('# Failed to commit offsets: %s: %s' % (err, partitions))
else:
Expand Down
2 changes: 1 addition & 1 deletion tests/test_Consumer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@ def test_basic_api():
except TypeError as e:
assert str(e) == "expected configuration dict"

def dummy_commit_cb (consumer, err, partitions):
def dummy_commit_cb (err, partitions):
pass

kc = Consumer({'group.id':'test', 'socket.timeout.ms':'100',
Expand Down