diff --git a/examples/kubernetes/producer/producer.py b/examples/kubernetes/producer/producer.py
index 317485c06..94fa5dbbb 100644
--- a/examples/kubernetes/producer/producer.py
+++ b/examples/kubernetes/producer/producer.py
@@ -1,20 +1,20 @@
 import json
 from random import random
-from kafka import KafkaProducer
+from aiokafka import KafkaProducer
 
-TOPIC = 'test'
-KEY = 'score'
+TOPIC = "test"
+KEY = "score"
 
 
 def publish_message(producer_instance, topic_name, key, value):
     try:
-        key_bytes = bytes(key, encoding='utf-8')
-        value_bytes = bytes(value, encoding='utf-8')
+        key_bytes = bytes(key, encoding="utf-8")
+        value_bytes = bytes(value, encoding="utf-8")
         producer_instance.send(topic_name, key=key_bytes, value=value_bytes)
         producer_instance.flush()
-        print('Message published successfully.')
+        print("Message published successfully.")
     except Exception as ex:
-        print('Exception in publishing message')
+        print("Exception in publishing message")
         print(ex)
 
 
@@ -26,20 +26,20 @@ def connect_kafka_producer():
         # Don't use in production, this only works with Docker for Mac in
         # development
         _producer = KafkaProducer(
-            bootstrap_servers=['host.docker.internal:9092'],
-            api_version=(0, 10))
+            bootstrap_servers=["host.docker.internal:9092"], api_version=(0, 10)
+        )
     except Exception as ex:
-        print('Exception while connecting Kafka')
+        print("Exception while connecting Kafka")
         print(ex)
     return _producer
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     kafka_producer = connect_kafka_producer()
     for index in range(0, 10000):
         message = {
-            'index': index,
-            'value': round(random(), 2),
+            "index": index,
+            "value": round(random(), 2),
         }
         publish_message(kafka_producer, TOPIC, KEY, json.dumps(message))
     if kafka_producer is not None:
diff --git a/faust/assignor/partition_assignor.py b/faust/assignor/partition_assignor.py
index 6711bbd01..705079620 100644
--- a/faust/assignor/partition_assignor.py
+++ b/faust/assignor/partition_assignor.py
@@ -4,9 +4,9 @@
 from collections import defaultdict
 from typing import Iterable, List, Mapping, MutableMapping, Sequence, Set, cast
 
-from kafka.cluster import ClusterMetadata
-from kafka.coordinator.assignors.abstract import AbstractPartitionAssignor
-from kafka.coordinator.protocol import (
+from aiokafka.cluster import ClusterMetadata
+from aiokafka.coordinator.assignors.abstract import AbstractPartitionAssignor
+from aiokafka.coordinator.protocol import (
     ConsumerProtocolMemberAssignment,
     ConsumerProtocolMemberMetadata,
 )
diff --git a/faust/tables/recovery.py b/faust/tables/recovery.py
index 9924880af..51ec3cb9c 100644
--- a/faust/tables/recovery.py
+++ b/faust/tables/recovery.py
@@ -18,7 +18,7 @@
 )
 
 import opentracing
-from kafka.errors import IllegalStateError
+from aiokafka.errors import IllegalStateError
 from mode import Service, get_logger
 from mode.services import WaitArgT
 from mode.utils.times import humanize_seconds, humanize_seconds_ago
diff --git a/faust/transport/drivers/aiokafka.py b/faust/transport/drivers/aiokafka.py
index 1630cd2c3..24a9de977 100644
--- a/faust/transport/drivers/aiokafka.py
+++ b/faust/transport/drivers/aiokafka.py
@@ -35,16 +35,16 @@
 )
 from aiokafka.structs import OffsetAndMetadata, TopicPartition as _TopicPartition
 from aiokafka.util import parse_kafka_version
-from kafka import TopicPartition
-from kafka.coordinator.assignors.roundrobin import RoundRobinPartitionAssignor
-from kafka.errors import (
+from aiokafka import TopicPartition
+from aiokafka.coordinator.assignors.roundrobin import RoundRobinPartitionAssignor
+from aiokafka.errors import (
     NotControllerError,
     TopicAlreadyExistsError as TopicExistsError,
     for_code,
 )
-from kafka.partitioner import murmur2
-from kafka.partitioner.default import DefaultPartitioner
-from kafka.protocol.metadata import MetadataRequest_v1
+from aiokafka.partitioner import murmur2
+from aiokafka.partitioner import DefaultPartitioner
+from aiokafka.protocol.metadata import MetadataRequest_v1
 from mode import Service, get_logger
 from mode.threads import ServiceThread, WorkerThread
 from mode.utils import text
diff --git a/faust/utils/kafka/protocol/admin.py b/faust/utils/kafka/protocol/admin.py
index b7d7a778f..3e418e960 100644
--- a/faust/utils/kafka/protocol/admin.py
+++ b/faust/utils/kafka/protocol/admin.py
@@ -1,5 +1,5 @@
 """Admin related Kafka protocol extensions."""
-from kafka.protocol import types
+from aiokafka.protocol import types
 
 from .api import Request, Response
 
diff --git a/faust/utils/kafka/protocol/api.py b/faust/utils/kafka/protocol/api.py
index 8e33bed57..523d9ef0f 100644
--- a/faust/utils/kafka/protocol/api.py
+++ b/faust/utils/kafka/protocol/api.py
@@ -3,8 +3,8 @@
 import abc
 from typing import Type
 
-from kafka.protocol.struct import Struct
-from kafka.protocol.types import Schema
+from aiokafka.protocol.struct import Struct
+from aiokafka.protocol.types import Schema
 
 
 class Response(Struct, metaclass=abc.ABCMeta):  # type: ignore
diff --git a/requirements/requirements.txt b/requirements/requirements.txt
index bf057344e..cdd50fedd 100644
--- a/requirements/requirements.txt
+++ b/requirements/requirements.txt
@@ -1,6 +1,6 @@
 aiohttp>=3.8.0,<4.0
 aiohttp_cors>=0.7,<2.0
-aiokafka>=0.8.0,<0.9.0
+aiokafka>=0.9.0
 click>=6.7,<8.2
 mode-streaming>=0.3.0
 opentracing>=1.3.0,<=2.4.0
diff --git a/tests/consistency/consistency_checker.py b/tests/consistency/consistency_checker.py
index d9b0f073c..76f73ea17 100644
--- a/tests/consistency/consistency_checker.py
+++ b/tests/consistency/consistency_checker.py
@@ -2,8 +2,8 @@
 from collections import defaultdict
 
 from aiokafka import AIOKafkaClient, AIOKafkaConsumer
-from kafka.protocol.commit import GroupCoordinatorRequest_v0, OffsetFetchRequest_v1
-from kafka.structs import TopicPartition
+from aiokafka.protocol.commit import GroupCoordinatorRequest_v0, OffsetFetchRequest_v1
+from aiokafka.structs import TopicPartition
 
 from faust.utils import json