aio-libs / aiokafka
Showing 10 of 35 files from the diff.

@@ -14,7 +14,7 @@
Loading
14 14
from aiokafka.record.memory_records import MemoryRecords
15 15
from aiokafka.record.control_record import ControlRecord, ABORT_MARKER
16 16
from aiokafka.structs import OffsetAndTimestamp, TopicPartition, ConsumerRecord
17 -
from aiokafka.util import ensure_future, create_future
17 +
from aiokafka.util import create_future, create_task
18 18
19 19
log = logging.getLogger(__name__)
20 20
@@ -58,18 +58,17 @@
Loading
58 58
59 59
class FetchResult:
60 60
    def __init__(
61 -
            self, tp, *, assignment, loop, partition_records, backoff):
61 +
            self, tp, *, assignment, partition_records, backoff):
62 62
        self._topic_partition = tp
63 63
        self._partition_records = partition_records
64 64
65 -
        self._created = loop.time()
65 +
        self._created = time.monotonic()
66 66
        self._backoff = backoff
67 -
        self._loop = loop
68 67
69 68
        self._assignment = assignment
70 69
71 70
    def calculate_backoff(self):
72 -
        lifetime = self._loop.time() - self._created
71 +
        lifetime = time.monotonic() - self._created
73 72
        if lifetime < self._backoff:
74 73
            return self._backoff - lifetime
75 74
        return 0
@@ -151,14 +150,13 @@
Loading
151 150
152 151
153 152
class FetchError:
154 -
    def __init__(self, *, loop, error, backoff):
153 +
    def __init__(self, *, error, backoff):
155 154
        self._error = error
156 -
        self._created = loop.time()
155 +
        self._created = time.monotonic()
157 156
        self._backoff = backoff
158 -
        self._loop = loop
159 157
160 158
    def calculate_backoff(self):
161 -
        lifetime = self._loop.time() - self._created
159 +
        lifetime = time.monotonic() - self._created
162 160
        if lifetime < self._backoff:
163 161
            return self._backoff - lifetime
164 162
        return 0
@@ -356,7 +354,7 @@
Loading
356 354
    """
357 355
358 356
    def __init__(
359 -
            self, client, subscriptions, *, loop,
357 +
            self, client, subscriptions, *,
360 358
            key_deserializer=None,
361 359
            value_deserializer=None,
362 360
            fetch_min_bytes=1,
@@ -370,7 +368,6 @@
Loading
370 368
            auto_offset_reset='latest',
371 369
            isolation_level="read_uncommitted"):
372 370
        self._client = client
373 -
        self._loop = loop
374 371
        self._key_deserializer = key_deserializer
375 372
        self._value_deserializer = value_deserializer
376 373
        self._fetch_min_bytes = fetch_min_bytes
@@ -414,8 +411,7 @@
Loading
414 411
            req_version = 1
415 412
        self._fetch_request_class = FetchRequest[req_version]
416 413
417 -
        self._fetch_task = ensure_future(
418 -
            self._fetch_requests_routine(), loop=loop)
414 +
        self._fetch_task = create_task(self._fetch_requests_routine())
419 415
420 416
        self._closed = False
421 417
@@ -444,7 +440,7 @@
Loading
444 440
        # Creating a fetch waiter is usually not that frequent of an operation,
445 441
        # (get methods will return all data first, before a waiter is created)
446 442
447 -
        fut = create_future(loop=self._loop)
443 +
        fut = create_future()
448 444
        self._fetch_waiters.add(fut)
449 445
        fut.add_done_callback(
450 446
            lambda f, waiters=self._fetch_waiters: waiters.remove(f))
@@ -475,7 +471,7 @@
Loading
475 471
            assignment = None
476 472
477 473
            def start_pending_task(coro, node_id, self=self):
478 -
                task = ensure_future(coro, loop=self._loop)
474 +
                task = create_task(coro)
479 475
                self._pending_tasks.add(task)
480 476
                self._in_flight.add(node_id)
481 477
@@ -510,7 +506,7 @@
Loading
510 506
                assert assignment is not None and assignment.active
511 507
512 508
                # Reset consuming signal future.
513 -
                self._wait_consume_future = create_future(loop=self._loop)
509 +
                self._wait_consume_future = create_future()
514 510
                # Determine what action to take per node
515 511
                (fetch_requests, reset_requests, timeout, invalid_metadata,
516 512
                 resume_futures) = self._get_actions_per_node(assignment)
@@ -535,7 +531,6 @@
Loading
535 531
536 532
                done_set, _ = await asyncio.wait(
537 533
                    chain(self._pending_tasks, other_futs, resume_futures),
538 -
                    loop=self._loop,
539 534
                    timeout=timeout,
540 535
                    return_when=asyncio.FIRST_COMPLETED)
541 536
@@ -659,7 +654,7 @@
Loading
659 654
            response = await self._client.send(node_id, request)
660 655
        except Errors.KafkaError as err:
661 656
            log.error("Failed fetch messages from %s: %s", node_id, err)
662 -
            await asyncio.sleep(self._retry_backoff, loop=self._loop)
657 +
            await asyncio.sleep(self._retry_backoff)
663 658
            return False
664 659
        except asyncio.CancelledError:
665 660
            # Either `close()` or partition unassigned. Either way the result
@@ -720,8 +715,7 @@
Loading
720 715
                        self._records[tp] = FetchResult(
721 716
                            tp, partition_records=partition_records,
722 717
                            assignment=assignment,
723 -
                            backoff=self._prefetch_backoff,
724 -
                            loop=self._loop)
718 +
                            backoff=self._prefetch_backoff)
725 719
726 720
                        # We added at least 1 successful record
727 721
                        needs_wakeup = True
@@ -768,7 +762,7 @@
Loading
768 762
    def _set_error(self, tp, error):
769 763
        assert tp not in self._records, self._records[tp]
770 764
        self._records[tp] = FetchError(
771 -
            error=error, backoff=self._prefetch_backoff, loop=self._loop)
765 +
            error=error, backoff=self._prefetch_backoff)
772 766
773 767
    async def _update_fetch_positions(self, assignment, node_id, tps):
774 768
        """ This task will be called if there is no valid position for
@@ -834,7 +828,7 @@
Loading
834 828
                    node_id, topic_data)
835 829
            except Errors.KafkaError as err:
836 830
                log.error("Failed fetch offsets from %s: %s", node_id, err)
837 -
                await asyncio.sleep(self._retry_backoff, loop=self._loop)
831 +
                await asyncio.sleep(self._retry_backoff)
838 832
                return needs_wakeup
839 833
        except asyncio.CancelledError:
840 834
            return needs_wakeup
@@ -869,14 +863,13 @@
Loading
869 863
            return {}
870 864
871 865
        timeout = timeout_ms / 1000
872 -
        start_time = self._loop.time()
866 +
        start_time = time.monotonic()
873 867
        remaining = timeout
874 868
        while True:
875 869
            try:
876 870
                offsets = await asyncio.wait_for(
877 871
                    self._proc_offset_requests(timestamps),
878 -
                    timeout=None if remaining == float("inf") else remaining,
879 -
                    loop=self._loop
872 +
                    timeout=None if remaining == float("inf") else remaining
880 873
                )
881 874
            except asyncio.TimeoutError:
882 875
                break
@@ -885,11 +878,11 @@
Loading
885 878
                    raise error
886 879
                if error.invalid_metadata:
887 880
                    self._client.force_metadata_update()
888 -
                elapsed = self._loop.time() - start_time
881 +
                elapsed = time.monotonic() - start_time
889 882
                remaining = max(0, remaining - elapsed)
890 883
                if remaining < self._retry_backoff:
891 884
                    break
892 -
                await asyncio.sleep(self._retry_backoff, loop=self._loop)
885 +
                await asyncio.sleep(self._retry_backoff)
893 886
            else:
894 887
                return offsets
895 888
        raise KafkaTimeoutError(
@@ -937,7 +930,7 @@
Loading
937 930
                self._proc_offset_request(node_id, topic_data)
938 931
            )
939 932
        offsets = {}
940 -
        res = await asyncio.gather(*futs, loop=self._loop)
933 +
        res = await asyncio.gather(*futs)
941 934
        for partial_offsets in res:
942 935
            offsets.update(partial_offsets)
943 936
        return offsets
@@ -1059,7 +1052,7 @@
Loading
1059 1052
            if self._subscriptions.reassignment_in_progress:
1060 1053
                await self._subscriptions.wait_for_assignment()
1061 1054
1062 -
            start_time = self._loop.time()
1055 +
            start_time = time.monotonic()
1063 1056
            drained = {}
1064 1057
            for tp in list(self._records.keys()):
1065 1058
                if partitions and tp not in partitions:
@@ -1097,8 +1090,7 @@
Loading
1097 1090
                return drained
1098 1091
1099 1092
            waiter = self._create_fetch_waiter()
1100 -
            done, pending = await asyncio.wait(
1101 -
                [waiter], timeout=timeout, loop=self._loop)
1093 +
            done, pending = await asyncio.wait([waiter], timeout=timeout)
1102 1094
1103 1095
            if not done or self._closed:
1104 1096
                if pending:
@@ -1110,7 +1102,7 @@
Loading
1110 1102
                waiter.result()  # Check for authorization errors
1111 1103
1112 1104
            # Decrease timeout accordingly
1113 -
            timeout = timeout - (self._loop.time() - start_time)
1105 +
            timeout = timeout - (time.monotonic() - start_time)
1114 1106
            timeout = max(0, timeout)
1115 1107
1116 1108
    async def get_offsets_by_times(self, timestamps, timeout_ms):
@@ -1159,7 +1151,7 @@
Loading
1159 1151
        # describing the purpose.
1160 1152
        self._notify(self._wait_consume_future)
1161 1153
1162 -
        return asyncio.gather(*waiters, loop=self._loop)
1154 +
        return asyncio.gather(*waiters)
1163 1155
1164 1156
    def seek_to(self, tp, offset):
1165 1157
        """ Force a position change to specific offset. Called from

@@ -7,6 +7,7 @@
Loading
7 7
import logging
8 8
import struct
9 9
import sys
10 +
import time
10 11
import traceback
11 12
import uuid
12 13
import warnings
@@ -20,7 +21,7 @@
Loading
20 21
    GroupCoordinatorResponse_v0 as GroupCoordinatorResponse)
21 22
22 23
import aiokafka.errors as Errors
23 -
from aiokafka.util import ensure_future, create_future
24 +
from aiokafka.util import create_future, create_task, get_running_loop
24 25
25 26
from aiokafka.abc import AbstractTokenProvider
26 27
@@ -68,7 +69,7 @@
Loading
68 69
69 70
70 71
async def create_conn(
71 -
    host, port, *, loop=None, client_id='aiokafka',
72 +
    host, port, *, client_id='aiokafka',
72 73
    request_timeout_ms=40000, api_version=(0, 8, 2),
73 74
    ssl_context=None, security_protocol="PLAINTEXT",
74 75
    max_idle_ms=None, on_close=None,
@@ -80,10 +81,8 @@
Loading
80 81
    sasl_oauth_token_provider=None,
81 82
    version_hint=None
82 83
):
83 -
    if loop is None:
84 -
        loop = asyncio.get_event_loop()
85 84
    conn = AIOKafkaConnection(
86 -
        host, port, loop=loop, client_id=client_id,
85 +
        host, port, client_id=client_id,
87 86
        request_timeout_ms=request_timeout_ms,
88 87
        api_version=api_version,
89 88
        ssl_context=ssl_context, security_protocol=security_protocol,
@@ -119,7 +118,7 @@
Loading
119 118
    _reader = None  # For __del__ to work properly, just in case
120 119
    _source_traceback = None
121 120
122 -
    def __init__(self, host, port, *, loop, client_id='aiokafka',
121 +
    def __init__(self, host, port, *, client_id='aiokafka',
123 122
                 request_timeout_ms=40000, api_version=(0, 8, 2),
124 123
                 ssl_context=None, security_protocol='PLAINTEXT',
125 124
                 max_idle_ms=None, on_close=None, sasl_mechanism=None,
@@ -128,6 +127,8 @@
Loading
128 127
                 sasl_kerberos_domain_name=None,
129 128
                 sasl_oauth_token_provider=None,
130 129
                 version_hint=None):
130 +
        loop = get_running_loop()
131 +
131 132
        if sasl_mechanism == "GSSAPI":
132 133
            assert gssapi is not None, "gssapi library required"
133 134
@@ -171,7 +172,7 @@
Loading
171 172
        self._closed_fut = None
172 173
173 174
        self._max_idle_ms = max_idle_ms
174 -
        self._last_action = loop.time()
175 +
        self._last_action = time.monotonic()
175 176
        self._idle_handle = None
176 177
177 178
        self._on_close_cb = on_close
@@ -202,7 +203,7 @@
Loading
202 203
203 204
    async def connect(self):
204 205
        loop = self._loop
205 -
        self._closed_fut = create_future(loop=loop)
206 +
        self._closed_fut = create_future()
206 207
        if self._security_protocol in ["PLAINTEXT", "SASL_PLAINTEXT"]:
207 208
            ssl = None
208 209
        else:
@@ -215,7 +216,7 @@
Loading
215 216
        transport, _ = await asyncio.wait_for(
216 217
            loop.create_connection(
217 218
                lambda: protocol, self.host, self.port, ssl=ssl),
218 -
            loop=loop, timeout=self._request_timeout)
219 +
            timeout=self._request_timeout)
219 220
        writer = asyncio.StreamWriter(transport, protocol, reader, loop)
220 221
        self._reader, self._writer, self._protocol = reader, writer, protocol
221 222
@@ -224,7 +225,7 @@
Loading
224 225
225 226
        # Start idle checker
226 227
        if self._max_idle_ms is not None:
227 -
            self._idle_handle = self._loop.call_soon(
228 +
            self._idle_handle = loop.call_soon(
228 229
                self._idle_check, weakref.ref(self))
229 230
230 231
        if self._version_hint and self._version_hint >= (0, 10):
@@ -383,7 +384,7 @@
Loading
383 384
    @staticmethod
384 385
    def _idle_check(self_ref):
385 386
        self = self_ref()
386 -
        idle_for = self._loop.time() - self._last_action
387 +
        idle_for = time.monotonic() - self._last_action
387 388
        timeout = self._max_idle_ms / 1000
388 389
        # If we have any pending requests, we are assumed to be not idle.
389 390
        # it's up to `request_timeout_ms` to break those.
@@ -435,9 +436,9 @@
Loading
435 436
436 437
        if not expect_response:
437 438
            return self._writer.drain()
438 -
        fut = create_future(loop=self._loop)
439 +
        fut = create_future()
439 440
        self._requests.append((correlation_id, request.RESPONSE_TYPE, fut))
440 -
        return asyncio.wait_for(fut, self._request_timeout, loop=self._loop)
441 +
        return asyncio.wait_for(fut, self._request_timeout)
441 442
442 443
    def _send_sasl_token(self, payload, expect_response=True):
443 444
        if self._writer is None:
@@ -457,9 +458,9 @@
Loading
457 458
        if not expect_response:
458 459
            return self._writer.drain()
459 460
460 -
        fut = create_future(loop=self._loop)
461 +
        fut = create_future()
461 462
        self._requests.append((None, None, fut))
462 -
        return asyncio.wait_for(fut, self._request_timeout, loop=self._loop)
463 +
        return asyncio.wait_for(fut, self._request_timeout)
463 464
464 465
    def connected(self):
465 466
        return bool(self._reader is not None and not self._reader.at_eof())
@@ -494,7 +495,7 @@
Loading
494 495
495 496
    def _create_reader_task(self):
496 497
        self_ref = weakref.ref(self)
497 -
        read_task = ensure_future(self._read(self_ref), loop=self._loop)
498 +
        read_task = create_task(self._read(self_ref))
498 499
        read_task.add_done_callback(
499 500
            functools.partial(self._on_read_task_error, self_ref))
500 501
        return read_task
@@ -549,7 +550,7 @@
Loading
549 550
                fut.set_result(response)
550 551
551 552
        # Update idle timer.
552 -
        self._last_action = self._loop.time()
553 +
        self._last_action = time.monotonic()
553 554
        # We should clear the request future only after all code is done and
554 555
        # future is resolved. If any fails it's up to close() method to fail
555 556
        # this future.

@@ -1,6 +1,7 @@
Loading
1 1
import asyncio
2 2
import collections
3 3
import logging
4 +
import time
4 5
5 6
import aiokafka.errors as Errors
6 7
from aiokafka.client import ConnectionGroup, CoordinationType
@@ -19,7 +20,7 @@
Loading
19 20
    AddOffsetsToTxnRequest, TxnOffsetCommitRequest
20 21
)
21 22
from aiokafka.structs import TopicPartition
22 -
from aiokafka.util import ensure_future
23 +
from aiokafka.util import create_task
23 24
24 25
log = logging.getLogger(__name__)
25 26
@@ -34,7 +35,7 @@
Loading
34 35
35 36
    def __init__(
36 37
            self, client, *, acks, txn_manager, message_accumulator,
37 -
            retry_backoff_ms, linger_ms, request_timeout_ms, loop):
38 +
            retry_backoff_ms, linger_ms, request_timeout_ms):
38 39
        self.client = client
39 40
        self._txn_manager = txn_manager
40 41
        self._acks = acks
@@ -44,7 +45,6 @@
Loading
44 45
        self._in_flight = set()
45 46
        self._muted_partitions = set()
46 47
        self._coordinators = {}
47 -
        self._loop = loop
48 48
        self._retry_backoff = retry_backoff_ms / 1000
49 49
        self._request_timeout_ms = request_timeout_ms
50 50
        self._linger_time = linger_ms / 1000
@@ -52,8 +52,7 @@
Loading
52 52
    async def start(self):
53 53
        # If producer is idempotent we need to assure we have PID found
54 54
        await self._maybe_wait_for_pid()
55 -
        self._sender_task = ensure_future(
56 -
            self._sender_routine(), loop=self._loop)
55 +
        self._sender_task = create_task(self._sender_routine())
57 56
        self._sender_task.add_done_callback(self._fail_all)
58 57
59 58
    def _fail_all(self, task):
@@ -121,9 +120,8 @@
Loading
121 120
122 121
                # create produce task for every batch
123 122
                for node_id, batches in batches.items():
124 -
                    task = ensure_future(
125 -
                        self._send_produce_req(node_id, batches),
126 -
                        loop=self._loop)
123 +
                    task = create_task(
124 +
                        self._send_produce_req(node_id, batches))
127 125
                    self._in_flight.add(node_id)
128 126
                    for tp in batches:
129 127
                        self._muted_partitions.add(tp)
@@ -144,8 +142,7 @@
Loading
144 142
                # * Metadata update if partition leader unknown
145 143
                done, _ = await asyncio.wait(
146 144
                    waiters,
147 -
                    return_when=asyncio.FIRST_COMPLETED,
148 -
                    loop=self._loop)
145 +
                    return_when=asyncio.FIRST_COMPLETED)
149 146
150 147
                # done tasks should never produce errors, if they are it's a
151 148
                # bug
@@ -204,7 +201,7 @@
Loading
204 201
                raise err
205 202
            except Errors.CoordinatorNotAvailableError:
206 203
                await self.client.force_metadata_update()
207 -
                await asyncio.sleep(self._retry_backoff, loop=self._loop)
204 +
                await asyncio.sleep(self._retry_backoff)
208 205
                continue
209 206
            except Errors.KafkaError as err:
210 207
                log.error("FindCoordinator Request failed: %s", err)
@@ -215,7 +212,7 @@
Loading
215 212
            ready = await self.client.ready(
216 213
                coordinator_id, group=ConnectionGroup.COORDINATION)
217 214
            if not ready:
218 -
                await asyncio.sleep(self._retry_backoff, loop=self._loop)
215 +
                await asyncio.sleep(self._retry_backoff)
219 216
                continue
220 217
221 218
            self._coordinators[coordinator_type] = coordinator_id
@@ -252,16 +249,16 @@
Loading
252 249
            node_id (int): kafka broker identifier
253 250
            batches (dict): dictionary of {TopicPartition: MessageBatch}
254 251
        """
255 -
        t0 = self._loop.time()
252 +
        t0 = time.monotonic()
256 253
257 254
        handler = SendProduceReqHandler(self, batches)
258 255
        await handler.do(node_id)
259 256
260 257
        # if batches for node is processed in less than a linger seconds
261 258
        # then waiting for the remaining time
262 -
        sleep_time = self._linger_time - (self._loop.time() - t0)
259 +
        sleep_time = self._linger_time - (time.monotonic() - t0)
263 260
        if sleep_time > 0:
264 -
            await asyncio.sleep(sleep_time, loop=self._loop)
261 +
            await asyncio.sleep(sleep_time)
265 262
266 263
        self._in_flight.remove(node_id)
267 264
        for tp in batches:
@@ -278,30 +275,26 @@
Loading
278 275
        # we need to do that before committing
279 276
        tps = txn_manager.partitions_to_add()
280 277
        if tps:
281 -
            return ensure_future(
282 -
                self._do_add_partitions_to_txn(tps),
283 -
                loop=self._loop)
278 +
            return create_task(
279 +
                self._do_add_partitions_to_txn(tps))
284 280
285 281
        # We need to add group to transaction before we can commit the offset
286 282
        group_id = txn_manager.consumer_group_to_add()
287 283
        if group_id is not None:
288 -
            return ensure_future(
289 -
                self._do_add_offsets_to_txn(group_id),
290 -
                loop=self._loop)
284 +
            return create_task(
285 +
                self._do_add_offsets_to_txn(group_id))
291 286
292 287
        # Now commit the added group's offset
293 288
        commit_data = txn_manager.offsets_to_commit()
294 289
        if commit_data is not None:
295 290
            offsets, group_id = commit_data
296 -
            return ensure_future(
297 -
                self._do_txn_offset_commit(offsets, group_id),
298 -
                loop=self._loop)
291 +
            return create_task(
292 +
                self._do_txn_offset_commit(offsets, group_id))
299 293
300 294
        commit_result = txn_manager.needs_transaction_commit()
301 295
        if commit_result is not None:
302 -
            return ensure_future(
303 -
                self._do_txn_commit(commit_result),
304 -
                loop=self._loop)
296 +
            return create_task(
297 +
                self._do_txn_commit(commit_result))
305 298
306 299
    async def _do_add_partitions_to_txn(self, tps):
307 300
        # First assert we have a valid coordinator to send the request to
@@ -368,7 +361,6 @@
Loading
368 361
    def __init__(self, sender):
369 362
        self._sender = sender
370 363
        self._default_backoff = sender._retry_backoff
371 -
        self._loop = sender._loop
372 364
373 365
    async def do(self, node_id):
374 366
        req = self.create_request()
@@ -377,12 +369,12 @@
Loading
377 369
                node_id, req, group=self.group)
378 370
        except KafkaError as err:
379 371
            log.warning("Could not send %r: %r", req.__class__, err)
380 -
            await asyncio.sleep(self._default_backoff, loop=self._loop)
372 +
            await asyncio.sleep(self._default_backoff)
381 373
            return False
382 374
383 375
        retry_backoff = self.handle_response(resp)
384 376
        if retry_backoff is not None:
385 -
            await asyncio.sleep(retry_backoff, loop=self._loop)
377 +
            await asyncio.sleep(retry_backoff)
386 378
            return False  # Failure
387 379
        else:
388 380
            return True  # Success
@@ -730,7 +722,7 @@
Loading
730 722
731 723
        if self._to_reenqueue:
732 724
            # Wait backoff before reequeue
733 -
            await asyncio.sleep(self._default_backoff, loop=self._loop)
725 +
            await asyncio.sleep(self._default_backoff)
734 726
735 727
            for batch in self._to_reenqueue:
736 728
                self._sender._message_accumulator.reenqueue(batch)

@@ -2,6 +2,7 @@
Loading
2 2
import collections
3 3
import logging
4 4
import copy
5 +
import time
5 6
6 7
from kafka.coordinator.assignors.roundrobin import RoundRobinPartitionAssignor
7 8
from kafka.coordinator.protocol import ConsumerProtocol
@@ -14,7 +15,7 @@
Loading
14 15
import aiokafka.errors as Errors
15 16
from aiokafka.structs import OffsetAndMetadata, TopicPartition
16 17
from aiokafka.client import ConnectionGroup, CoordinationType
17 -
from aiokafka.util import ensure_future, create_future
18 +
from aiokafka.util import create_future, create_task
18 19
19 20
log = logging.getLogger(__name__)
20 21
@@ -23,9 +24,8 @@
Loading
23 24
24 25
class BaseCoordinator(object):
25 26
26 -
    def __init__(self, client, subscription, *, loop,
27 +
    def __init__(self, client, subscription, *,
27 28
                 exclude_internal_topics=True):
28 -
        self._loop = loop
29 29
        self._client = client
30 30
        self._exclude_internal_topics = exclude_internal_topics
31 31
        self._subscription = subscription
@@ -85,8 +85,8 @@
Loading
85 85
    def __init__(self, *args, **kw):
86 86
        super().__init__(*args, **kw)
87 87
        # Reset all committed points, as the GroupCoordinator would
88 -
        self._reset_committed_task = ensure_future(
89 -
            self._reset_committed_routine(), loop=self._loop)
88 +
        self._reset_committed_task = create_task(
89 +
            self._reset_committed_routine())
90 90
91 91
    def _on_metadata_change(self):
92 92
        self.assign_all_partitions()
@@ -141,13 +141,11 @@
Loading
141 141
                    tp_state.update_committed(
142 142
                        OffsetAndMetadata(UNKNOWN_OFFSET, ""))
143 143
144 -
                event_waiter = ensure_future(
145 -
                    commit_refresh_needed.wait(), loop=self._loop)
144 +
                event_waiter = create_task(commit_refresh_needed.wait())
146 145
147 146
                await asyncio.wait(
148 147
                    [assignment.unassign_future, event_waiter],
149 -
                    return_when=asyncio.FIRST_COMPLETED,
150 -
                    loop=self._loop)
148 +
                    return_when=asyncio.FIRST_COMPLETED)
151 149
152 150
                if not event_waiter.done():
153 151
                    event_waiter.cancel()
@@ -210,7 +208,7 @@
Loading
210 208
          org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
211 209
    """
212 210
213 -
    def __init__(self, client, subscription, *, loop,
211 +
    def __init__(self, client, subscription, *,
214 212
                 group_id='aiokafka-default-group',
215 213
                 session_timeout_ms=10000, heartbeat_interval_ms=3000,
216 214
                 retry_backoff_ms=100,
@@ -228,7 +226,7 @@
Loading
228 226
        self._group_subscription = None
229 227
230 228
        super().__init__(
231 -
            client, subscription, loop=loop,
229 +
            client, subscription,
232 230
            exclude_internal_topics=exclude_internal_topics)
233 231
234 232
        self._session_timeout_ms = session_timeout_ms
@@ -247,11 +245,10 @@
Loading
247 245
248 246
        # Coordination flags and futures
249 247
        self._performed_join_prepare = False
250 -
        self._rejoin_needed_fut = create_future(loop=loop)
251 -
        self._coordinator_dead_fut = create_future(loop=loop)
248 +
        self._rejoin_needed_fut = create_future()
249 +
        self._coordinator_dead_fut = create_future()
252 250
253 -
        self._coordination_task = ensure_future(
254 -
            self._coordination_routine(), loop=loop)
251 +
        self._coordination_task = create_task(self._coordination_routine())
255 252
256 253
        # Will be started/stopped by coordination task
257 254
        self._heartbeat_task = None
@@ -263,16 +260,16 @@
Loading
263 260
        self._pending_exception = None
264 261
        self._error_consumed_fut = None
265 262
266 -
        self._coordinator_lookup_lock = asyncio.Lock(loop=loop)
263 +
        self._coordinator_lookup_lock = asyncio.Lock()
267 264
        # Will synchronize edits to TopicPartitionState.committed, as it may be
268 265
        # changed from user code by calling ``commit()``.
269 -
        self._commit_lock = asyncio.Lock(loop=loop)
266 +
        self._commit_lock = asyncio.Lock()
270 267
271 268
        self._next_autocommit_deadline = \
272 -
            loop.time() + auto_commit_interval_ms / 1000
269 +
            time.monotonic() + auto_commit_interval_ms / 1000
273 270
274 271
        # Will be set on close
275 -
        self._closing = create_future(loop=loop)
272 +
        self._closing = create_future()
276 273
277 274
    def _on_metadata_change(self):
278 275
        self.request_rejoin()
@@ -323,11 +320,10 @@
Loading
323 320
        exc = copy.copy(exc)
324 321
        self._subscription.abort_waiters(exc)
325 322
        self._pending_exception = exc
326 -
        self._error_consumed_fut = create_future(loop=self._loop)
323 +
        self._error_consumed_fut = create_future()
327 324
        return asyncio.wait(
328 325
            [self._error_consumed_fut, self._closing],
329 326
            return_when=asyncio.FIRST_COMPLETED,
330 -
            loop=self._loop
331 327
        )
332 328
333 329
    async def close(self):
@@ -346,7 +342,7 @@
Loading
346 342
        await self._maybe_leave_group()
347 343
348 344
    def maybe_leave_group(self):
349 -
        task = ensure_future(self._maybe_leave_group(), loop=self._loop)
345 +
        task = create_task(self._maybe_leave_group())
350 346
        return task
351 347
352 348
    async def _maybe_leave_group(self):
@@ -536,8 +532,7 @@
Loading
536 532
                    log.error("Group Coordinator Request failed: %s", err)
537 533
                    if err.retriable:
538 534
                        await self._client.force_metadata_update()
539 -
                        await asyncio.sleep(
540 -
                            retry_backoff, loop=self._loop)
535 +
                        await asyncio.sleep(retry_backoff)
541 536
                        continue
542 537
                    else:
543 538
                        raise
@@ -547,11 +542,11 @@
Loading
547 542
                ready = await self._client.ready(
548 543
                    coordinator_id, group=ConnectionGroup.COORDINATION)
549 544
                if not ready:
550 -
                    await asyncio.sleep(retry_backoff, loop=self._loop)
545 +
                    await asyncio.sleep(retry_backoff)
551 546
                    continue
552 547
553 548
                self.coordinator_id = coordinator_id
554 -
                self._coordinator_dead_fut = create_future(loop=self._loop)
549 +
                self._coordinator_dead_fut = create_future()
555 550
                log.info("Discovered coordinator %s for group %s",
556 551
                         self.coordinator_id, self.group_id)
557 552
@@ -589,7 +584,7 @@
Loading
589 584
                await asyncio.wait(
590 585
                    [self._subscription.wait_for_subscription(),
591 586
                     self._closing],
592 -
                    return_when=asyncio.FIRST_COMPLETED, loop=self._loop)
587 +
                    return_when=asyncio.FIRST_COMPLETED)
593 588
                if self._closing.done():
594 589
                    break
595 590
                subscription = self._subscription.subscription
@@ -641,7 +636,7 @@
Loading
641 636
                futures.append(self._commit_refresh_task)
642 637
643 638
            done, _ = await asyncio.wait(
644 -
                futures, timeout=wait_timeout, loop=self._loop,
639 +
                futures, timeout=wait_timeout,
645 640
                return_when=asyncio.FIRST_COMPLETED)
646 641
647 642
            # Handle exceptions in other background tasks
@@ -706,8 +701,7 @@
Loading
706 701
707 702
    def _start_heartbeat_task(self):
708 703
        if self._heartbeat_task is None:
709 -
            self._heartbeat_task = ensure_future(
710 -
                self._heartbeat_routine(), loop=self._loop)
704 +
            self._heartbeat_task = create_task(self._heartbeat_routine())
711 705
712 706
    async def _stop_heartbeat_task(self):
713 707
        if self._heartbeat_task is not None:
@@ -717,7 +711,7 @@
Loading
717 711
            self._heartbeat_task = None
718 712
719 713
    async def _heartbeat_routine(self):
720 -
        last_ok_heartbeat = self._loop.time()
714 +
        last_ok_heartbeat = time.monotonic()
721 715
        hb_interval = self._heartbeat_interval_ms / 1000
722 716
        session_timeout = self._session_timeout_ms / 1000
723 717
        retry_backoff = self._retry_backoff_ms / 1000
@@ -727,10 +721,10 @@
Loading
727 721
        # this consumer, so we stop after resetting generation.
728 722
        while self.member_id != JoinGroupRequest[0].UNKNOWN_MEMBER_ID:
729 723
            try:
730 -
                await asyncio.sleep(sleep_time, loop=self._loop)
724 +
                await asyncio.sleep(sleep_time)
731 725
                await self.ensure_coordinator_known()
732 726
733 -
                t0 = self._loop.time()
727 +
                t0 = time.monotonic()
734 728
                success = await self._do_heartbeat()
735 729
            except asyncio.CancelledError:
736 730
                break
@@ -739,12 +733,12 @@
Loading
739 733
            # routine
740 734
741 735
            if success:
742 -
                last_ok_heartbeat = self._loop.time()
743 -
                sleep_time = max((0, hb_interval - self._loop.time() + t0))
736 +
                last_ok_heartbeat = time.monotonic()
737 +
                sleep_time = max((0, hb_interval - last_ok_heartbeat + t0))
744 738
            else:
745 739
                sleep_time = retry_backoff
746 740
747 -
            session_time = self._loop.time() - last_ok_heartbeat
741 +
            session_time = time.monotonic() - last_ok_heartbeat
748 742
            if session_time > session_timeout:
749 743
                # the session timeout has expired without seeing a successful
750 744
                # heartbeat, so we should probably make sure the coordinator
@@ -827,8 +821,8 @@
Loading
827 821
    def start_commit_offsets_refresh_task(self, assignment):
828 822
        if self._commit_refresh_task is not None:
829 823
            self._commit_refresh_task.cancel()
830 -
        self._commit_refresh_task = ensure_future(
831 -
            self._commit_refresh_routine(assignment), loop=self._loop)
824 +
        self._commit_refresh_task = create_task(
825 +
            self._commit_refresh_routine(assignment))
832 826
833 827
    async def _stop_commit_offsets_refresh_task(self):
834 828
        # The previous task should end after assignment changed
@@ -856,15 +850,14 @@
Loading
856 850
                    timeout = retry_backoff_ms
857 851
                else:
858 852
                    timeout = None
859 -
                    event_waiter = ensure_future(
860 -
                        commit_refresh_needed.wait(), loop=self._loop)
853 +
                    event_waiter = create_task(
854 +
                        commit_refresh_needed.wait())
861 855
                    wait_futures.append(event_waiter)
862 856
863 857
                await asyncio.wait(
864 858
                    wait_futures,
865 859
                    timeout=timeout,
866 -
                    return_when=asyncio.FIRST_COMPLETED,
867 -
                    loop=self._loop)
860 +
                    return_when=asyncio.FIRST_COMPLETED)
868 861
        except asyncio.CancelledError:
869 862
            pass
870 863
        except Exception:
@@ -882,7 +875,7 @@
Loading
882 875
        rebalance = CoordinatorGroupRebalance(
883 876
            self, self.group_id, self.coordinator_id,
884 877
            subscription, self._assignors, self._session_timeout_ms,
885 -
            self._retry_backoff_ms, loop=self._loop)
878 +
            self._retry_backoff_ms)
886 879
        assignment = await rebalance.perform_group_join()
887 880
888 881
        if not subscription.active:
@@ -894,7 +887,7 @@
Loading
894 887
        if assignment is None:
895 888
            # wait backoff and try again
896 889
            await asyncio.sleep(
897 -
                self._retry_backoff_ms / 1000, loop=self._loop)
890 +
                self._retry_backoff_ms / 1000)
898 891
            return False
899 892
900 893
        protocol, member_assignment_bytes = assignment
@@ -906,7 +899,7 @@
Loading
906 899
    async def _maybe_do_autocommit(self, assignment):
907 900
        if not self._enable_auto_commit:
908 901
            return None
909 -
        now = self._loop.time()
902 +
        now = time.monotonic()
910 903
        interval = self._auto_commit_interval_ms / 1000
911 904
        backoff = self._retry_backoff_ms / 1000
912 905
        if now > self._next_autocommit_deadline:
@@ -919,7 +912,7 @@
Loading
919 912
                if self._is_commit_retriable(error):
920 913
                    # Retry after backoff.
921 914
                    self._next_autocommit_deadline = \
922 -
                        self._loop.time() + backoff
915 +
                        time.monotonic() + backoff
923 916
                    return backoff
924 917
                else:
925 918
                    raise
@@ -927,7 +920,7 @@
Loading
927 920
            # from another source (say Fetcher, like authorization errors).
928 921
            self._next_autocommit_deadline = now + interval
929 922
930 -
        return max(0, self._next_autocommit_deadline - self._loop.time())
923 +
        return max(0, self._next_autocommit_deadline - time.monotonic())
931 924
932 925
    def _is_commit_retriable(self, error):
933 926
        # Java client raises CommitFailedError which is retriable and thus
@@ -958,8 +951,7 @@
Loading
958 951
            try:
959 952
                async with self._commit_lock:
960 953
                    await asyncio.shield(
961 -
                        self._do_commit_offsets(assignment, offsets),
962 -
                        loop=self._loop)
954 +
                        self._do_commit_offsets(assignment, offsets))
963 955
            except (Errors.UnknownMemberIdError,
964 956
                    Errors.IllegalGenerationError,
965 957
                    Errors.RebalanceInProgressError):
@@ -973,7 +965,7 @@
Loading
973 965
                else:
974 966
                    # wait backoff and try again
975 967
                    await asyncio.sleep(
976 -
                        self._retry_backoff_ms / 1000, loop=self._loop)
968 +
                        self._retry_backoff_ms / 1000)
977 969
            else:
978 970
                break
979 971
@@ -1120,7 +1112,7 @@
Loading
1120 1112
                else:
1121 1113
                    # wait backoff and try again
1122 1114
                    await asyncio.sleep(
1123 -
                        self._retry_backoff_ms / 1000, loop=self._loop)
1115 +
                        self._retry_backoff_ms / 1000)
1124 1116
            else:
1125 1117
                return offsets
1126 1118
@@ -1180,11 +1172,10 @@
Loading
1180 1172
    """
1181 1173
1182 1174
    def __init__(self, coordinator, group_id, coordinator_id, subscription,
1183 -
                 assignors, session_timeout_ms, retry_backoff_ms, *, loop):
1175 +
                 assignors, session_timeout_ms, retry_backoff_ms):
1184 1176
        self._coordinator = coordinator
1185 1177
        self.group_id = group_id
1186 1178
        self.coordinator_id = coordinator_id
1187 -
        self._loop = loop
1188 1179
1189 1180
        self._subscription = subscription
1190 1181
        self._assignors = assignors
@@ -1276,7 +1267,7 @@
Loading
1276 1267
                      " is loading the group.", self.group_id,
1277 1268
                      self.coordinator_id)
1278 1269
            await asyncio.sleep(
1279 -
                self._retry_backoff_ms / 1000, loop=self._loop)
1270 +
                self._retry_backoff_ms / 1000)
1280 1271
        elif error_type is Errors.UnknownMemberIdError:
1281 1272
            # reset the member id and retry immediately
1282 1273
            self._coordinator.reset_generation()
@@ -1364,7 +1355,7 @@
Loading
1364 1355
        # capture metadata changes after join group was performed. We do not
1365 1356
        # set it directly after JoinGroup to avoid a false rejoin in case
1366 1357
        # ``_perform_assignment()`` does a metadata update.
1367 -
        self._coordinator._rejoin_needed_fut = create_future(loop=self._loop)
1358 +
        self._coordinator._rejoin_needed_fut = create_future()
1368 1359
        try:
1369 1360
            response = await self._coordinator._send_req(request)
1370 1361
        except Errors.KafkaError:

@@ -1,6 +1,7 @@
Loading
1 1
import asyncio
2 2
import logging
3 3
import random
4 +
import time
4 5
5 6
from kafka.conn import collect_hosts
6 7
from kafka.protocol.metadata import MetadataRequest
@@ -22,7 +23,7 @@
Loading
22 23
    UnrecognizedBrokerVersion,
23 24
    StaleMetadata)
24 25
from aiokafka.util import (
25 -
    ensure_future, create_future, get_running_loop, parse_kafka_version
26 +
    create_task, create_future, parse_kafka_version, get_running_loop
26 27
)
27 28
28 29
@@ -147,8 +148,8 @@
Loading
147 148
        self._sync_task = None
148 149
149 150
        self._md_update_fut = None
150 -
        self._md_update_waiter = create_future(loop=self._loop)
151 -
        self._get_conn_lock = asyncio.Lock(loop=loop)
151 +
        self._md_update_waiter = create_future()
152 +
        self._get_conn_lock = asyncio.Lock()
152 153
153 154
    def __repr__(self):
154 155
        return '<AIOKafkaClient client_id=%s>' % self._client_id
@@ -178,10 +179,13 @@
Loading
178 179
        for conn in self._conns.values():
179 180
            futs.append(conn.close(reason=CloseReason.SHUTDOWN))
180 181
        if futs:
181 -
            await asyncio.gather(*futs, loop=self._loop)
182 +
            await asyncio.gather(*futs)
182 183
183 184
    async def bootstrap(self):
184 185
        """Try to to bootstrap initial cluster metadata"""
186 +
        assert self._loop is asyncio.get_event_loop(), (
187 +
            "Please create objects with the same loop as running with"
188 +
        )
185 189
        # using request v0 for bootstrap if not sure v1 is available
186 190
        if self._api_version == "auto" or self._api_version < (0, 10):
187 191
            metadata_request = MetadataRequest[0]([])
@@ -197,7 +201,7 @@
Loading
197 201
198 202
            try:
199 203
                bootstrap_conn = await create_conn(
200 -
                    host, port, loop=self._loop, client_id=self._client_id,
204 +
                    host, port, client_id=self._client_id,
201 205
                    request_timeout_ms=self._request_timeout_ms,
202 206
                    ssl_context=self._ssl_context,
203 207
                    security_protocol=self._security_protocol,
@@ -244,8 +248,7 @@
Loading
244 248
245 249
        if self._sync_task is None:
246 250
            # starting metadata synchronizer task
247 -
            self._sync_task = ensure_future(
248 -
                self._md_synchronizer(), loop=self._loop)
251 +
            self._sync_task = create_task(self._md_synchronizer())
249 252
250 253
    async def _md_synchronizer(self):
251 254
        """routine (async task) for synchronize cluster metadata every
@@ -253,12 +256,11 @@
Loading
253 256
        while True:
254 257
            await asyncio.wait(
255 258
                [self._md_update_waiter],
256 -
                timeout=self._metadata_max_age_ms / 1000,
257 -
                loop=self._loop)
259 +
                timeout=self._metadata_max_age_ms / 1000)
258 260
259 261
            topics = self._topics
260 262
            if self._md_update_fut is None:
261 -
                self._md_update_fut = create_future(loop=self._loop)
263 +
                self._md_update_fut = create_future()
262 264
            ret = await self._metadata_update(self.cluster, topics)
263 265
            # If list of topics changed during metadata update we must update
264 266
            # it again right away.
@@ -267,7 +269,7 @@
Loading
267 269
            # Earlier this waiter was set before sending metadata_request,
268 270
            # but that was to avoid topic list changes being unnoticed, which
269 271
            # is handled explicitly now.
270 -
            self._md_update_waiter = create_future(loop=self._loop)
272 +
            self._md_update_waiter = create_future()
271 273
272 274
            self._md_update_fut.set_result(ret)
273 275
            self._md_update_fut = None
@@ -342,9 +344,9 @@
Loading
342 344
            # Wake up the `_md_synchronizer` task
343 345
            if not self._md_update_waiter.done():
344 346
                self._md_update_waiter.set_result(None)
345 -
            self._md_update_fut = create_future(loop=self._loop)
347 +
            self._md_update_fut = create_future()
346 348
        # Metadata will be updated in the background by syncronizer
347 -
        return asyncio.shield(self._md_update_fut, loop=self._loop)
349 +
        return asyncio.shield(self._md_update_fut)
348 350
349 351
    async def fetch_all_metadata(self):
350 352
        cluster_md = ClusterMetadata(
@@ -362,7 +364,7 @@
Loading
362 364
            topic (str): topic to track
363 365
        """
364 366
        if topic in self._topics:
365 -
            res = create_future(loop=self._loop)
367 +
            res = create_future()
366 368
            res.set_result(True)
367 369
        else:
368 370
            res = self.force_metadata_update()
@@ -379,7 +381,7 @@
Loading
379 381
        if not topics or set(topics).difference(self._topics):
380 382
            res = self.force_metadata_update()
381 383
        else:
382 -
            res = create_future(loop=self._loop)
384 +
            res = create_future()
383 385
            res.set_result(True)
384 386
        self._topics = set(topics)
385 387
        return res
@@ -432,7 +434,7 @@
Loading
432 434
                    version_hint = None
433 435
434 436
                self._conns[conn_id] = await create_conn(
435 -
                    broker.host, broker.port, loop=self._loop,
437 +
                    broker.host, broker.port,
436 438
                    client_id=self._client_id,
437 439
                    request_timeout_ms=self._request_timeout_ms,
438 440
                    ssl_context=self._ssl_context,
@@ -544,8 +546,8 @@
Loading
544 546
                assert conn, 'no connection to node with id {}'.format(node_id)
545 547
                # request can be ignored by Kafka broker,
546 548
                # so we send metadata request and wait response
547 -
                task = self._loop.create_task(conn.send(request))
548 -
                await asyncio.wait([task], timeout=0.1, loop=self._loop)
549 +
                task = create_task(conn.send(request))
550 +
                await asyncio.wait([task], timeout=0.1)
549 551
                try:
550 552
                    await conn.send(MetadataRequest_v0([]))
551 553
                except KafkaError:
@@ -617,23 +619,22 @@
Loading
617 619
        # add topic to metadata topic list if it is not there already.
618 620
        self.add_topic(topic)
619 621
620 -
        t0 = self._loop.time()
622 +
        t0 = time.monotonic()
621 623
        while True:
622 624
            await self.force_metadata_update()
623 625
            if topic in self.cluster.topics():
624 626
                break
625 -
            if (self._loop.time() - t0) > (self._request_timeout_ms / 1000):
627 +
            if (time.monotonic() - t0) > (self._request_timeout_ms / 1000):
626 628
                raise UnknownTopicOrPartitionError()
627 629
            if topic in self.cluster.unauthorized_topics:
628 630
                raise Errors.TopicAuthorizationFailedError(topic)
629 -
            await asyncio.sleep(self._retry_backoff, loop=self._loop)
631 +
            await asyncio.sleep(self._retry_backoff)
630 632
631 633
        return self.cluster.partitions_for_topic(topic)
632 634
633 635
    async def _maybe_wait_metadata(self):
634 636
        if self._md_update_fut is not None:
635 -
            await asyncio.shield(
636 -
                self._md_update_fut, loop=self._loop)
637 +
            await asyncio.shield(self._md_update_fut)
637 638
638 639
    async def coordinator_lookup(self, coordinator_type, coordinator_key):
639 640
        """ Lookup which node in the cluster is the coordinator for a certain

@@ -1,7 +1,8 @@
Loading
1 1
import logging
2 2
import contextlib
3 3
import copy
4 -
from asyncio import AbstractEventLoop as ALoop, shield, Event, Future
4 +
import time
5 +
from asyncio import shield, Event, Future
5 6
from enum import Enum
6 7
7 8
from typing import Set, Pattern, Dict, List
@@ -41,14 +42,13 @@
Loading
41 42
    _subscription = None  # type: Subscription
42 43
    _listener = None  # type: ConsumerRebalanceListener
43 44
44 -
    def __init__(self, *, loop: ALoop):
45 +
    def __init__(self):
45 46
        self._subscription_waiters = []  # type: List[Future]
46 47
        self._assignment_waiters = []  # type: List[Future]
47 -
        self._loop = loop  # type: ALoop
48 48
49 49
        # Fetch contexts
50 50
        self._fetch_count = 0
51 -
        self._last_fetch_ended = loop.time()
51 +
        self._last_fetch_ended = time.monotonic()
52 52
53 53
    @property
54 54
    def subscription(self) -> "Subscription":
@@ -145,7 +145,7 @@
Loading
145 145
                isinstance(listener, ConsumerRebalanceListener))
146 146
        self._set_subscription_type(SubscriptionType.AUTO_TOPICS)
147 147
148 -
        self._change_subscription(Subscription(topics, loop=self._loop))
148 +
        self._change_subscription(Subscription(topics))
149 149
        self._listener = listener
150 150
        self._notify_subscription_waiters()
151 151
@@ -175,7 +175,7 @@
Loading
175 175
        self._set_subscription_type(SubscriptionType.USER_ASSIGNED)
176 176
177 177
        self._change_subscription(
178 -
            ManualSubscription(partitions, loop=self._loop))
178 +
            ManualSubscription(partitions))
179 179
        self._notify_assignment_waiters()
180 180
181 181
    def unsubscribe(self):
@@ -204,7 +204,7 @@
Loading
204 204
        Affects: SubscriptionState.subscription
205 205
        """
206 206
        assert self._subscription_type == SubscriptionType.AUTO_PATTERN
207 -
        self._change_subscription(Subscription(topics, loop=self._loop))
207 +
        self._change_subscription(Subscription(topics))
208 208
209 209
    def assign_from_subscribed(self, assignment: Set[TopicPartition]):
210 210
        """ Set assignment if automatic assignment is used.
@@ -244,7 +244,7 @@
Loading
244 244
        """ Wait for subscription change. This will always wait for next
245 245
        subscription.
246 246
        """
247 -
        fut = create_future(loop=self._loop)
247 +
        fut = create_future()
248 248
        self._subscription_waiters.append(fut)
249 249
        return fut
250 250
@@ -252,7 +252,7 @@
Loading
252 252
        """ Wait for next assignment. Be careful, as this will always wait for
253 253
        next assignment, even if the current one is active.
254 254
        """
255 -
        fut = create_future(loop=self._loop)
255 +
        fut = create_future()
256 256
        self._assignment_waiters.append(fut)
257 257
        return fut
258 258
@@ -294,13 +294,13 @@
Loading
294 294
        yield
295 295
        self._fetch_count -= 1
296 296
        if self._fetch_count == 0:
297 -
            self._last_fetch_ended = self._loop.time()
297 +
            self._last_fetch_ended = time.monotonic()
298 298
299 299
    @property
300 300
    def fetcher_idle_time(self):
301 301
        """ How much time (in seconds) spent without consuming any records """
302 302
        if self._fetch_count == 0:
303 -
            return self._loop.time() - self._last_fetch_ended
303 +
            return time.monotonic() - self._last_fetch_ended
304 304
        else:
305 305
            return 0
306 306
@@ -316,11 +316,10 @@
Loading
316 316
        * Unsubscribed
317 317
    """
318 318
319 -
    def __init__(self, topics: Set[str], *, loop: ALoop):
319 +
    def __init__(self, topics: Set[str]):
320 320
        self._topics = frozenset(topics)  # type: Set[str]
321 321
        self._assignment = None  # type: Assignment
322 -
        self._loop = loop  # type: ALoop
323 -
        self.unsubscribe_future = create_future(loop)  # type: Future
322 +
        self.unsubscribe_future = create_future()  # type: Future
324 323
        self._reassignment_in_progress = True
325 324
326 325
    @property
@@ -343,7 +342,7 @@
Loading
343 342
        if self._assignment is not None:
344 343
            self._assignment._unassign()
345 344
346 -
        self._assignment = Assignment(topic_partitions, loop=self._loop)
345 +
        self._assignment = Assignment(topic_partitions)
347 346
        self._reassignment_in_progress = False
348 347
349 348
    def _unsubscribe(self):
@@ -359,15 +358,14 @@
Loading
359 358
    """ Describes a user assignment
360 359
    """
361 360
362 -
    def __init__(self, user_assignment: Set[TopicPartition], *, loop):
361 +
    def __init__(self, user_assignment: Set[TopicPartition]):
363 362
        topics = set([])
364 363
        for tp in user_assignment:
365 364
            topics.add(tp.topic)
366 365
367 366
        self._topics = frozenset(topics)
368 -
        self._assignment = Assignment(user_assignment, loop=loop)
369 -
        self._loop = loop
370 -
        self.unsubscribe_future = create_future(loop)
367 +
        self._assignment = Assignment(user_assignment)
368 +
        self.unsubscribe_future = create_future()
371 369
372 370
    def _assign(
373 371
            self, topic_partitions: Set[TopicPartition]):  # pragma: no cover
@@ -390,18 +388,17 @@
Loading
390 388
        * Unassigned
391 389
    """
392 390
393 -
    def __init__(self, topic_partitions: Set[TopicPartition], *, loop):
391 +
    def __init__(self, topic_partitions: Set[TopicPartition]):
394 392
        assert isinstance(topic_partitions, (list, set, tuple))
395 393
396 394
        self._topic_partitions = frozenset(topic_partitions)
397 395
398 396
        self._tp_state = {}  # type: Dict[TopicPartition, TopicPartitionState]
399 397
        for tp in self._topic_partitions:
400 -
            self._tp_state[tp] = TopicPartitionState(self, loop=loop)
398 +
            self._tp_state[tp] = TopicPartitionState(self)
401 399
402 -
        self._loop = loop
403 -
        self.unassign_future = create_future(loop)
404 -
        self.commit_refresh_needed = Event(loop=loop)
400 +
        self.unassign_future = create_future()
401 +
        self.commit_refresh_needed = Event()
405 402
406 403
    @property
407 404
    def tps(self):
@@ -457,7 +454,7 @@
Loading
457 454
458 455
    """
459 456
460 -
    def __init__(self, assignment, *, loop):
457 +
    def __init__(self, assignment):
461 458
        # Synchronized values
462 459
        self._committed_futs = []
463 460
@@ -465,7 +462,7 @@
Loading
465 462
        self.lso = None  # Last fetched stable offset mark
466 463
        self.timestamp = None  # timestamp of last poll
467 464
        self._position = None  # The current position of the topic
468 -
        self._position_fut = create_future(loop=loop)
465 +
        self._position_fut = create_future()
469 466
470 467
        # Will be set by `seek_to_beginning` or `seek_to_end` if called by user
471 468
        # or by Fetcher after confirming that current position is no longer
@@ -473,7 +470,6 @@
Loading
473 470
        self._reset_strategy = None  # type: int
474 471
        self._status = PartitionStatus.AWAITING_RESET  # type: PartitionStatus
475 472
476 -
        self._loop = loop
477 473
        self._assignment = assignment
478 474
479 475
        self._paused = False
@@ -511,13 +507,13 @@
Loading
511 507
        self._reset_strategy = strategy
512 508
        self._position = None
513 509
        if self._position_fut.done():
514 -
            self._position_fut = create_future(loop=self._loop)
510 +
            self._position_fut = create_future()
515 511
        self._status = PartitionStatus.AWAITING_RESET
516 512
517 513
    # Committed manipulation
518 514
519 515
    def fetch_committed(self):
520 -
        fut = create_future(loop=self._loop)
516 +
        fut = create_future()
521 517
        self._committed_futs.append(fut)
522 518
        self._assignment.commit_refresh_needed.set()
523 519
        return fut
@@ -559,14 +555,14 @@
Loading
559 555
            self._position_fut.set_result(None)
560 556
561 557
    def wait_for_position(self):
562 -
        return shield(self._position_fut, loop=self._loop)
558 +
        return shield(self._position_fut)
563 559
564 560
    # Pause/Unpause
565 561
    def pause(self):
566 562
        if not self._paused:
567 563
            self._paused = True
568 564
            assert self._resume_fut is None
569 -
            self._resume_fut = create_future(loop=self._loop)
565 +
            self._resume_fut = create_future()
570 566
571 567
    def resume(self):
572 568
        if self._paused:

@@ -218,7 +218,7 @@
Loading
218 218
                    "acks={} not supported if enable_idempotence=True"
219 219
                    .format(acks))
220 220
            self._txn_manager = TransactionManager(
221 -
                transactional_id, transaction_timeout_ms, loop=loop)
221 +
                transactional_id, transaction_timeout_ms)
222 222
        else:
223 223
            self._txn_manager = None
224 224
@@ -256,14 +256,12 @@
Loading
256 256
        self._metadata = self.client.cluster
257 257
        self._message_accumulator = MessageAccumulator(
258 258
            self._metadata, max_batch_size, compression_attrs,
259 -
            self._request_timeout_ms / 1000, txn_manager=self._txn_manager,
260 -
            loop=loop)
259 +
            self._request_timeout_ms / 1000, txn_manager=self._txn_manager)
261 260
        self._sender = Sender(
262 261
            self.client, acks=acks, txn_manager=self._txn_manager,
263 262
            retry_backoff_ms=retry_backoff_ms, linger_ms=linger_ms,
264 263
            message_accumulator=self._message_accumulator,
265 -
            request_timeout_ms=request_timeout_ms,
266 -
            loop=loop)
264 +
            request_timeout_ms=request_timeout_ms)
267 265
268 266
        self._loop = loop
269 267
        if loop.get_debug():
@@ -285,6 +283,9 @@
Loading
285 283
286 284
    async def start(self):
287 285
        """Connect to Kafka cluster and check server version"""
286 +
        assert self._loop is asyncio.get_event_loop(), (
287 +
            "Please create objects with the same loop as running with"
288 +
        )
288 289
        log.debug("Starting the Kafka producer")  # trace
289 290
        await self.client.bootstrap()
290 291
@@ -317,8 +318,7 @@
Loading
317 318
            await asyncio.wait([
318 319
                self._message_accumulator.close(),
319 320
                self._sender.sender_task],
320 -
                return_when=asyncio.FIRST_COMPLETED,
321 -
                loop=self._loop)
321 +
                return_when=asyncio.FIRST_COMPLETED)
322 322
323 323
            await self._sender.close()
324 324
@@ -509,8 +509,7 @@
Loading
509 509
            "Beginning a new transaction for id %s",
510 510
            self._txn_manager.transactional_id)
511 511
        await asyncio.shield(
512 -
            self._txn_manager.wait_for_pid(),
513 -
            loop=self._loop
512 +
            self._txn_manager.wait_for_pid()
514 513
        )
515 514
        self._txn_manager.begin_transaction()
516 515
@@ -522,7 +521,6 @@
Loading
522 521
        self._txn_manager.committing_transaction()
523 522
        await asyncio.shield(
524 523
            self._txn_manager.wait_for_transaction_end(),
525 -
            loop=self._loop
526 524
        )
527 525
528 526
    async def abort_transaction(self):
@@ -533,7 +531,6 @@
Loading
533 531
        self._txn_manager.aborting_transaction()
534 532
        await asyncio.shield(
535 533
            self._txn_manager.wait_for_transaction_end(),
536 -
            loop=self._loop
537 534
        )
538 535
539 536
    def transaction(self):
@@ -555,7 +552,7 @@
Loading
555 552
            "Begin adding offsets %s for consumer group %s to transaction",
556 553
            formatted_offsets, group_id)
557 554
        fut = self._txn_manager.add_offsets_to_txn(formatted_offsets, group_id)
558 -
        await asyncio.shield(fut, loop=self._loop)
555 +
        await asyncio.shield(fut)
559 556
560 557
    async def __aenter__(self):
561 558
        await self.start()

@@ -53,13 +53,13 @@
Loading
53 53
54 54
class TransactionManager:
55 55
56 -
    def __init__(self, transactional_id, transaction_timeout_ms, *, loop):
56 +
    def __init__(self, transactional_id, transaction_timeout_ms):
57 57
        self.transactional_id = transactional_id
58 58
        self.transaction_timeout_ms = transaction_timeout_ms
59 59
        self.state = TransactionState.UNINITIALIZED
60 60
61 61
        self._pid_and_epoch = PidAndEpoch(NO_PRODUCER_ID, NO_PRODUCER_EPOCH)
62 -
        self._pid_waiter = create_future(loop)
62 +
        self._pid_waiter = create_future()
63 63
        self._sequence_numbers = defaultdict(lambda: 0)
64 64
        self._transaction_waiter = None
65 65
        self._task_waiter = None
@@ -69,8 +69,6 @@
Loading
69 69
        self._txn_consumer_group = None
70 70
        self._pending_txn_offsets = deque()
71 71
72 -
        self._loop = loop
73 -
74 72
    # INDEMPOTANCE PART
75 73
76 74
    def set_pid_and_epoch(self, pid: int, epoch: int):
@@ -116,7 +114,7 @@
Loading
116 114
117 115
    def begin_transaction(self):
118 116
        self._transition_to(TransactionState.IN_TRANSACTION)
119 -
        self._transaction_waiter = create_future(loop=self._loop)
117 +
        self._transaction_waiter = create_future()
120 118
121 119
    def committing_transaction(self):
122 120
        if self.state == TransactionState.ABORTABLE_ERROR:
@@ -131,7 +129,7 @@
Loading
131 129
132 130
        # If we had an abortable error we need to create a new waiter
133 131
        if self._transaction_waiter.done():
134 -
            self._transaction_waiter = create_future(loop=self._loop)
132 +
            self._transaction_waiter = create_future()
135 133
        self.notify_task_waiter()
136 134
137 135
    def complete_transaction(self):
@@ -163,7 +161,7 @@
Loading
163 161
        self._pending_txn_offsets.clear()
164 162
        # There may be an abortable error. We just override it
165 163
        if self._transaction_waiter.done():
166 -
            self._transaction_waiter = create_future(loop=self._loop)
164 +
            self._transaction_waiter = create_future()
167 165
        self._transaction_waiter.set_exception(exc)
168 166
169 167
    def maybe_add_partition_to_txn(self, tp: TopicPartition):
@@ -177,7 +175,7 @@
Loading
177 175
    def add_offsets_to_txn(self, offsets, group_id):
178 176
        assert self.is_in_transaction()
179 177
        assert self.transactional_id
180 -
        fut = create_future(loop=self._loop)
178 +
        fut = create_future()
181 179
        self._pending_txn_offsets.append(
182 180
            (group_id, offsets, fut)
183 181
        )
@@ -249,5 +247,5 @@
Loading
249 247
            self._task_waiter.set_result(None)
250 248
251 249
    def make_task_waiter(self):
252 -
        self._task_waiter = create_future(loop=self._loop)
250 +
        self._task_waiter = create_future()
253 251
        return self._task_waiter

@@ -302,7 +302,7 @@
Loading
302 302
        self._max_poll_interval_ms = max_poll_interval_ms
303 303
304 304
        self._check_crcs = check_crcs
305 -
        self._subscription = SubscriptionState(loop=loop)
305 +
        self._subscription = SubscriptionState()
306 306
        self._fetcher = None
307 307
        self._coordinator = None
308 308
        self._loop = loop
@@ -334,6 +334,9 @@
Loading
334 334
            * Wait for possible topic autocreation
335 335
            * Join group if ``group_id`` provided
336 336
        """
337 +
        assert self._loop is asyncio.get_event_loop(), (
338 +
            "Please create objects with the same loop as running with"
339 +
        )
337 340
        assert self._fetcher is None, "Did you call `start` twice?"
338 341
        await self._client.bootstrap()
339 342
        await self._wait_topics()
@@ -349,7 +352,7 @@
Loading
349 352
                "0.11 and above")
350 353
351 354
        self._fetcher = Fetcher(
352 -
            self._client, self._subscription, loop=self._loop,
355 +
            self._client, self._subscription,
353 356
            key_deserializer=self._key_deserializer,
354 357
            value_deserializer=self._value_deserializer,
355 358
            fetch_min_bytes=self._fetch_min_bytes,
@@ -365,7 +368,7 @@
Loading
365 368
        if self._group_id is not None:
366 369
            # using group coordinator for automatic partitions assignment
367 370
            self._coordinator = GroupCoordinator(
368 -
                self._client, self._subscription, loop=self._loop,
371 +
                self._client, self._subscription,
369 372
                group_id=self._group_id,
370 373
                heartbeat_interval_ms=self._heartbeat_interval_ms,
371 374
                session_timeout_ms=self._session_timeout_ms,
@@ -392,7 +395,7 @@
Loading
392 395
            # Using a simple assignment coordinator for reassignment on
393 396
            # metadata changes
394 397
            self._coordinator = NoGroupCoordinator(
395 -
                self._client, self._subscription, loop=self._loop,
398 +
                self._client, self._subscription,
396 399
                exclude_internal_topics=self._exclude_internal_topics)
397 400
398 401
            if self._subscription.subscription is not None:
@@ -633,7 +636,7 @@
Loading
633 636
                    [tp_state.wait_for_position(),
634 637
                     assignment.unassign_future],
635 638
                    timeout=self._request_timeout_ms / 1000,
636 -
                    return_when=asyncio.FIRST_COMPLETED, loop=self._loop,
639 +
                    return_when=asyncio.FIRST_COMPLETED
637 640
                )
638 641
                if not tp_state.has_valid_position:
639 642
                    if self._subscription.subscription is None:
@@ -776,8 +779,7 @@
Loading
776 779
        await asyncio.wait(
777 780
            [fut, assignment.unassign_future],
778 781
            timeout=self._request_timeout_ms / 1000,
779 -
            return_when=asyncio.FIRST_COMPLETED,
780 -
            loop=self._loop
782 +
            return_when=asyncio.FIRST_COMPLETED
781 783
        )
782 784
        self._coordinator.check_errors()
783 785
        return fut.done()
@@ -818,8 +820,7 @@
Loading
818 820
        await asyncio.wait(
819 821
            [fut, assignment.unassign_future],
820 822
            timeout=self._request_timeout_ms / 1000,
821 -
            return_when=asyncio.FIRST_COMPLETED,
822 -
            loop=self._loop
823 +
            return_when=asyncio.FIRST_COMPLETED
823 824
        )
824 825
        self._coordinator.check_errors()
825 826
        return fut.done()

@@ -1,6 +1,7 @@
Loading
1 1
import asyncio
2 2
import collections
3 3
import copy
4 +
import time
4 5
5 6
from aiokafka.errors import (KafkaTimeoutError,
6 7
                             NotLeaderForPartitionError,
@@ -102,19 +103,18 @@
Loading
102 103
class MessageBatch:
103 104
    """This class incapsulate operations with batch of produce messages"""
104 105
105 -
    def __init__(self, tp, builder, ttl, loop):
106 +
    def __init__(self, tp, builder, ttl):
106 107
        self._builder = builder
107 108
        self._tp = tp
108 -
        self._loop = loop
109 109
        self._ttl = ttl
110 -
        self._ctime = loop.time()
110 +
        self._ctime = time.monotonic()
111 111
112 112
        # Waiters
113 113
        # Set when messages are delivered to Kafka based on ACK setting
114 -
        self.future = create_future(loop)
114 +
        self.future = create_future()
115 115
        self._msg_futures = []
116 116
        # Set when sender takes this batch
117 -
        self._drain_waiter = create_future(loop=loop)
117 +
        self._drain_waiter = create_future()
118 118
        self._retry_count = 0
119 119
120 120
    @property
@@ -139,7 +139,7 @@
Loading
139 139
        if metadata is None:
140 140
            return None
141 141
142 -
        future = _create_future(loop=self._loop)
142 +
        future = _create_future()
143 143
        self._msg_futures.append((future, metadata))
144 144
        return future
145 145
@@ -203,19 +203,18 @@
Loading
203 203
204 204
    def wait_deliver(self, timeout=None):
205 205
        """Wait until all message from this batch is processed"""
206 -
        return asyncio.wait([self.future], timeout=timeout, loop=self._loop)
206 +
        return asyncio.wait([self.future], timeout=timeout)
207 207
208 208
    async def wait_drain(self, timeout=None):
209 209
        """Wait until all message from this batch is processed"""
210 210
        waiter = self._drain_waiter
211 -
        await asyncio.wait(
212 -
            [waiter], timeout=timeout, loop=self._loop)
211 +
        await asyncio.wait([waiter], timeout=timeout)
213 212
        if waiter.done():
214 213
            waiter.result()  # Check for exception
215 214
216 215
    def expired(self):
217 216
        """Check that batch is expired or not"""
218 -
        return (self._loop.time() - self._ctime) > self._ttl
217 +
        return (time.monotonic() - self._ctime) > self._ttl
219 218
220 219
    def drain_ready(self):
221 220
        """Compress batch to be ready for send"""
@@ -226,7 +225,7 @@
Loading
226 225
    def reset_drain(self):
227 226
        """Reset drain waiter, until we will do another retry"""
228 227
        assert self._drain_waiter.done()
229 -
        self._drain_waiter = create_future(self._loop)
228 +
        self._drain_waiter = create_future()
230 229
231 230
    def set_producer_state(self, producer_id, producer_epoch, base_sequence):
232 231
        assert not self._drain_waiter.done()
@@ -252,15 +251,14 @@
Loading
252 251
    """
253 252
    def __init__(
254 253
            self, cluster, batch_size, compression_type, batch_ttl, *,
255 -
            txn_manager=None, loop):
254 +
            txn_manager=None):
256 255
        self._batches = collections.defaultdict(collections.deque)
257 256
        self._pending_batches = set([])
258 257
        self._cluster = cluster
259 258
        self._batch_size = batch_size
260 259
        self._compression_type = compression_type
261 260
        self._batch_ttl = batch_ttl
262 -
        self._loop = loop
263 -
        self._wait_data_future = create_future(loop=loop)
261 +
        self._wait_data_future = create_future()
264 262
        self._closed = False
265 263
        self._api_version = (0, 9)
266 264
        self._txn_manager = txn_manager
@@ -292,7 +290,7 @@
Loading
292 290
        # above, other batches should not be delivered as part of this
293 291
        # transaction
294 292
        if waiters:
295 -
            await asyncio.wait(waiters, loop=self._loop)
293 +
            await asyncio.wait(waiters)
296 294
297 295
    def fail_all(self, exception):
298 296
        # Close all batches with this exception
@@ -335,9 +333,9 @@
Loading
335 333
                return future
336 334
            # Batch is full, can't append data atm,
337 335
            # waiting until batch per topic-partition is drained
338 -
            start = self._loop.time()
336 +
            start = time.monotonic()
339 337
            await batch.wait_drain(timeout)
340 -
            timeout -= self._loop.time() - start
338 +
            timeout -= time.monotonic() - start
341 339
            if timeout <= 0:
342 340
                raise KafkaTimeoutError()
343 341
@@ -417,7 +415,7 @@
Loading
417 415
        # task
418 416
        if not self._wait_data_future.done():
419 417
            self._wait_data_future.set_result(None)
420 -
        self._wait_data_future = create_future(loop=self._loop)
418 +
        self._wait_data_future = create_future()
421 419
422 420
        return nodes, unknown_leaders_exist
423 421
@@ -442,7 +440,7 @@
Loading
442 440
        if self._txn_manager is not None:
443 441
            self._txn_manager.maybe_add_partition_to_txn(tp)
444 442
445 -
        batch = MessageBatch(tp, builder, self._batch_ttl, self._loop)
443 +
        batch = MessageBatch(tp, builder, self._batch_ttl)
446 444
        self._batches[tp].append(batch)
447 445
        if not self._wait_data_future.done():
448 446
            self._wait_data_future.set_result(None)
@@ -471,13 +469,13 @@
Loading
471 469
        if self._exception is not None:
472 470
            raise copy.copy(self._exception)
473 471
474 -
        start = self._loop.time()
472 +
        start = time.monotonic()
475 473
        while timeout > 0:
476 474
            pending = self._batches.get(tp)
477 475
            if pending:
478 476
                await pending[-1].wait_drain(timeout=timeout)
479 -
                timeout -= self._loop.time() - start
477 +
                timeout -= time.monotonic() - start
480 478
            else:
481 479
                batch = self._append_batch(builder, tp)
482 -
                return asyncio.shield(batch.future, loop=self._loop)
480 +
                return asyncio.shield(batch.future)
483 481
        raise KafkaTimeoutError()
Files Coverage
aiokafka 97.96%
Project Totals (29 files) 97.96%
test-linux-3.8.5-2.2.2-2.12-py
Build #320478060 -
integration purepy
test-linux-3.6-2.4.0-2.12-py
Build #320478060 -
integration purepy
test-linux-3.8.5-2.2.2-2.12-ext
Build #320478060 -
integration cext
test-linux-3.8.5-2.3.1-2.12-py
Build #331358311 -
integration purepy
test-linux-3.7-2.4.0-2.12-ext
Build #331358311 -
integration cext
test-linux-3.7-2.4.0-2.12-py
Build #331358311 -
integration purepy
test-linux-3.6-2.4.0-2.12-ext
Build #331358311 -
integration cext
test-linux-3.6-2.4.0-2.12-py
Build #331358311 -
integration purepy
test-linux-3.8.5-2.2.2-2.12-ext
Build #331358311 -
integration cext
test-linux-3.8.5-2.2.2-2.12-py
Build #331358311 -
integration purepy
test-linux-3.8.5-0.11.0.3-2.12-ext
Build #331358311 -
integration cext
test-linux-3.8.5-0.11.0.3-2.12-py
Build #331358311 -
integration purepy
test-linux-3.8.5-2.4.0-2.12-ext
Build #331358311 -
integration cext
m6ego125q3dt4emj
test-linux-3.8.5-1.1.1-2.12-ext
Build #320478060 -
integration cext
test-linux-3.8.5-1.1.1-2.12-py
Build #320478060 -
integration purepy
test-linux-3.8.5-2.1.1-2.12-ext
Build #320478060 -
integration cext
test-linux-3.6-2.4.0-2.12-ext
Build #320478060 -
integration cext
test-linux-3.8.5-0.9.0.1-2.11-ext
Build #320478060 -
integration cext
test-linux-3.8.5-0.9.0.1-2.11-py
Build #320478060 -
integration purepy
test-windows-3.8.5-py
Build #331358311 -
unit purepy
test-mac-3.7-ext
Build #331358311 -
unit cext
test-mac-3.8.5-py
Build #331358311 -
unit purepy
test-linux-3.8.5-0.10.2.1-2.11-py
Build #320478060 -
integration purepy
test-windows-3.7-py
Build #331358311 -
unit purepy
test-mac-3.8.5-ext
Build #331358311 -
unit cext
test-windows-3.6-py
Build #331358311 -
unit purepy
test-windows-3.7-ext
Build #331358311 -
unit cext
skwmwxvsccysfwge
test-windows-3.6-ext
Build #331358311 -
unit cext
ityevg79s4udomk8
ityevg79s4udomk8
test-mac-3.7-py
Build #320478060 -
unit purepy
test-mac-3.7-ext
Build #320478060 -
unit cext
w8anwrmhbws49rnp
w8anwrmhbws49rnp
test-mac-3.8.5-py
Build #320478060 -
unit purepy
test-mac-3.8.5-ext
Build #320478060 -
unit cext
test-windows-3.7-py
Build #320478060 -
unit purepy
test-windows-3.7-ext
Build #320478060 -
unit cext
8tsh4l8grgihl9hk
p9b2lyooolfxufy9
yk34f1x5gt0y3gr5
yk34f1x5gt0y3gr5
test-linux-3.8.5-2.1.1-2.12-py
Build #331358311 -
integration purepy
test-linux-3.8.5-2.1.1-2.12-ext
Build #331358311 -
integration cext
test-linux-3.8.5-2.4.0-2.12-py
Build #331358311 -
integration purepy
test-windows-3.8.5-py
Build #320478060 -
unit purepy
test-linux-3.8.5-2.3.1-2.12-py
Build #320478060 -
integration purepy
test-linux-3.8.5-1.1.1-2.12-py
Build #331358311 -
integration purepy
test-windows-3.6-py
Build #320478060 -
unit purepy
test-windows-3.6-ext
Build #320478060 -
unit cext
skwmwxvsccysfwge
cb3kidd90siiof74
test-linux-3.8.5-2.3.1-2.12-ext
Build #331358311 -
integration cext
test-linux-3.8.5-0.10.2.1-2.11-ext
Build #320478060 -
integration cext
test-windows-3.8.5-ext
Build #320478060 -
unit cext
test-mac-3.6-ext
Build #320478060 -
unit cext
8tsh4l8grgihl9hk
test-linux-3.8.5-0.9.0.1-2.11-py
Build #331358311 -
integration purepy
test-linux-3.8.5-2.3.1-2.12-ext
Build #320478060 -
integration cext
test-linux-3.8.5-0.10.2.1-2.11-py
Build #331358311 -
integration purepy
test-mac-3.6-ext
Build #331358311 -
unit cext
p9b2lyooolfxufy9
test-linux-3.8.5-0.9.0.1-2.11-ext
Build #331358311 -
integration cext
test-mac-3.6-py
Build #331358311 -
unit purepy
m6ego125q3dt4emj
test-linux-3.8.5-0.10.2.1-2.11-ext
Build #331358311 -
integration cext
cb3kidd90siiof74
test-windows-3.8.5-ext
Build #331358311 -
unit cext
test-mac-3.6-py
Build #320478060 -
unit purepy
test-linux-3.8.5-1.1.1-2.12-ext
Build #331358311 -
integration cext
test-mac-3.7-py
Build #331358311 -
unit purepy
1
codecov:
2
  token: f98ce702-d158-4f21-bdf2-d98dd7c92ca2
Sunburst
The inner-most circle is the entire project, moving away from the center are folders then, finally, a single file. The size and color of each slice is representing the number of statements and the coverage, respectively.
Icicle
The top section represents the entire project. Proceeding with folders and finally individual files. The size and color of each slice is representing the number of statements and the coverage, respectively.