to_kafka throughput
Nobody has claimed this yet.
Assessment
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Newbie friendliness
- 35/100
- Issue type
- Bug
- Clarity
- Mostly clear
- Activity status
- Stale
- Tech stack
- kafka, python
- Domain
- stream-processing
Research direction
Start at the to_kafka sink and its self.producer.poll(0) call, then reproduce the behavior with the proposed test_to_kafka_throughput test. Done means 100 emitted items reach sink_to_list() within the test timeout without callbacks being limited by the 0.2-second poll interval.
Written by the indexing model from the issue text.
Description
I'm testing to_kafka sink and its throughput is limited by polltime (0.2 sec). Looks like self.producer.poll(0) only polls for one message at a time and so only one callback is called every 0.2 seconds.
This fails:
def test_to_kafka_throughput():
ARGS = {'bootstrap.servers': 'localhost:9092'}
with kafka_service() as kafka:
_, TOPIC = kafka
source = Stream.from_iterable(range(100)).map(lambda x: str(x).encode())
kafka = source.to_kafka(TOPIC, ARGS)
out = kafka.sink_to_list()
source.start()
wait_for(
lambda: len(out) == 100,
5,
period=0.1,
fail_func=lambda: print("len(out) ==", len(out))
)
The existing test_to_kafka test doesn't catch this, because it starts waiting on the result only after all the items are emitted.
I spent some time tinkering with the code, but can't figure out what's wrong and how to fix this, so any ideas are appreciated.
- Dominant language
- Python
- Stars
- 1.3k
- Forks
- 149
- PR merge metrics
- No merged PRs in 30d
Contributor guide
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- Open a pull request that references the issue number.
More from python-streamz/streamz
-
Difficulty 3/5 1-2 days Newbie friendliness 35/100
python-streamz/streamz#481 · 4 comments ·
-
Combining the streamz.Stream.filenames() and streamz.Stream.from_textfile() using dask scatter? Open
Difficulty 5/5 Over a week Newbie friendliness 25/100
python-streamz/streamz#480 · 2 comments ·
-
Difficulty 5/5 Over a week Newbie friendliness 20/100
python-streamz/streamz#479 · 1 comment ·
-
Difficulty 5/5 Over a week Newbie friendliness 20/100
python-streamz/streamz#478 · 6 comments ·
-
Difficulty 4/5 3-5 days Newbie friendliness 10/100
python-streamz/streamz#476 · 17 comments · 2 reactions ·
All issues in python-streamz/streamz
Similar issues
-
Difficulty 2/5 1-3 hours Newbie friendliness 88/100
-
Difficulty 2/5 1-3 hours Newbie friendliness 82/100
-
Difficulty 2/5 1-3 hours Newbie friendliness 78/100
-
enhancement
Difficulty 2/5 1-3 hours Newbie friendliness 72/100
-
Difficulty 2/5 1-3 hours Newbie friendliness 74/100