diff --git a/kafka/coordinator/base.py b/kafka/coordinator/base.py
index 168115a..25dd000 100644
--- a/kafka/coordinator/base.py
+++ b/kafka/coordinator/base.py
@@ -50,6 +50,7 @@ class BaseCoordinator(object):
         'session_timeout_ms': 30000,
         'heartbeat_interval_ms': 3000,
         'retry_backoff_ms': 100,
+        'api_version': (0, 9),
     }
 
     def __init__(self, client, **configs):
@@ -194,6 +195,14 @@ class BaseCoordinator(object):
         """
         while self.coordinator_unknown():
 
+            # Prior to 0.8.2 there was no group coordinator
+            # so we will just pick a node at random and treat
+            # it as the "coordinator"
+            if self.config['api_version'] < (0, 8, 2):
+                self.coordinator_id = self._client.least_loaded_node()
+                self._client.ready(self.coordinator_id)
+                continue
+
             future = self._send_group_coordinator_request()
             self._client.poll(future=future)
 
diff --git a/kafka/coordinator/consumer.py b/kafka/coordinator/consumer.py
index 196bcc7..083a36a 100644
--- a/kafka/coordinator/consumer.py
+++ b/kafka/coordinator/consumer.py
@@ -100,6 +100,12 @@ class ConsumerCoordinator(BaseCoordinator):
                 interval = self.config['auto_commit_interval_ms'] / 1000.0
                 self._auto_commit_task = AutoCommitTask(weakref.proxy(self), interval)
 
+                # When using broker-coordinated consumer groups, auto-commit will
+                # be automatically enabled on group join (see _on_join_complete)
+                # Otherwise, we should enable now b/c there will be no group join
+                if self.config['api_version'] < (0, 9):
+                    self._auto_commit_task.enable()
+
         self._sensors = ConsumerCoordinatorMetrics(metrics, metric_group_prefix,
                                                    self._subscription)
 
@@ -293,8 +299,7 @@ class ConsumerCoordinator(BaseCoordinator):
             return {}
 
         while True:
-            if self.config['api_version'] >= (0, 8, 2):
-                self.ensure_coordinator_known()
+            self.ensure_coordinator_known()
 
             # contact coordinator to fetch committed offsets
             future = self._send_offset_fetch_request(partitions)
@@ -356,8 +361,7 @@ class ConsumerCoordinator(BaseCoordinator):
             return
 
         while True:
-            if self.config['api_version'] >= (0, 8, 2):
-                self.ensure_coordinator_known()
+            self.ensure_coordinator_known()
 
             future = self._send_offset_commit_request(offsets)
             self._client.poll(future=future)
@@ -415,14 +419,10 @@ class ConsumerCoordinator(BaseCoordinator):
             log.debug('No offsets to commit')
             return Future().success(True)
 
-        if self.config['api_version'] >= (0, 8, 2):
-            if self.coordinator_unknown():
-                return Future().failure(Errors.GroupCoordinatorNotAvailableError)
-            node_id = self.coordinator_id
-        else:
-            node_id = self._client.least_loaded_node()
-            if node_id is None:
-                return Future().failure(Errors.NoBrokersAvailable)
+        elif self.coordinator_unknown():
+            return Future().failure(Errors.GroupCoordinatorNotAvailableError)
+
+        node_id = self.coordinator_id
 
         # create the offset commit request
         offset_data = collections.defaultdict(dict)
@@ -571,14 +571,10 @@ class ConsumerCoordinator(BaseCoordinator):
         if not partitions:
             return Future().success({})
 
-        if self.config['api_version'] >= (0, 8, 2):
-            if self.coordinator_unknown():
-                return Future().failure(Errors.GroupCoordinatorNotAvailableError)
-            node_id = self.coordinator_id
-        else:
-            node_id = self._client.least_loaded_node()
-            if node_id is None:
-                return Future().failure(Errors.NoBrokersAvailable)
+        elif self.coordinator_unknown():
+            return Future().failure(Errors.GroupCoordinatorNotAvailableError)
+
+        node_id = self.coordinator_id
 
         # Verify node is ready
         if not self._client.ready(node_id):

Support KafkaConsumer auto-commit with 0.8 brokers
kafka 0.8.2    kafka-python 1.1.1

when enable auto_commit, an AutoCommitTask instance will be created
but when Enable the AutoCommitTask instance?
in the code , only find the function _on_join_complete will enable  the AutoCommitTask instance

```
def _on_join_complete(self, generation, member_id, protocol,
                          member_assignment_bytes):
        ...
        # restart the autocommit task if needed
        if self._auto_commit_task:
            self._auto_commit_task.enable()
```

but api_version < 0.9, the function _on_join_complete never be called
diff --git a/test/test_coordinator.py b/test/test_coordinator.py
index 15b915d..735d278 100644
--- a/test/test_coordinator.py
+++ b/test/test_coordinator.py
@@ -425,8 +425,7 @@ def test_send_offset_commit_request_fail(patched_coord, offsets):
     ((0, 9), OffsetCommitRequest[2])])
 def test_send_offset_commit_request_versions(patched_coord, offsets,
                                              api_version, req_type):
-    # assuming fixture sets coordinator=0, least_loaded_node=1
-    expect_node = 0 if api_version >= (0, 8, 2) else 1
+    expect_node = 0
     patched_coord.config['api_version'] = api_version
 
     patched_coord._send_offset_commit_request(offsets)
@@ -522,7 +521,7 @@ def test_send_offset_fetch_request_fail(patched_coord, partitions):
 def test_send_offset_fetch_request_versions(patched_coord, partitions,
                                             api_version, req_type):
     # assuming fixture sets coordinator=0, least_loaded_node=1
-    expect_node = 0 if api_version >= (0, 8, 2) else 1
+    expect_node = 0
     patched_coord.config['api_version'] = api_version
 
     patched_coord._send_offset_fetch_request(partitions)

pip install -e .
parse_log_pytest
pytest
pytest pytest-cov pytest-catchlog pytest-sugar pytest-mock mock python-snappy lz4tools xxhash six
apt-get update apt-get install -y gcc libsnappy-dev
3.5
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
attrs @ file:///opt/conda/conda-bld/attrs_1642510447205/work
certifi==2021.5.30
coverage==6.2
cramjam==2.5.0
importlib-metadata @ file:///tmp/build/80754af9/importlib-metadata_1631916693255/work
iniconfig @ file:///home/linux1/recipes/ci/iniconfig_1610983019677/work
-e git+https://github.com/dpkp/kafka-python.git@7a350e5fcf33f49094c820ba88b9cee4aeae6e12#egg=kafka_python
lz4tools==1.3.1.2
mock==5.2.0
more-itertools @ file:///tmp/build/80754af9/more-itertools_1637733554872/work
packaging @ file:///tmp/build/80754af9/packaging_1637314298585/work
pluggy @ file:///tmp/build/80754af9/pluggy_1615976315926/work
py @ file:///opt/conda/conda-bld/py_1644396412707/work
pyparsing @ file:///tmp/build/80754af9/pyparsing_1635766073266/work
pytest==6.2.4
pytest-catchlog==1.2.2
pytest-cov==4.0.0
pytest-mock==3.6.1
pytest-sugar==0.9.6
python-snappy==0.7.3
six==1.17.0
termcolor==1.1.0
toml @ file:///tmp/build/80754af9/toml_1616166611790/work
tomli==1.2.3
typing_extensions @ file:///opt/conda/conda-bld/typing_extensions_1647553014482/work
xxhash==3.2.0
zipp @ file:///tmp/build/80754af9/zipp_1633618647012/work

name: kafka-python
channels:
  - defaults
  - https://repo.anaconda.com/pkgs/main
  - https://repo.anaconda.com/pkgs/r
  - conda-forge
dependencies:
  - _libgcc_mutex=0.1=main
  - _openmp_mutex=5.1=1_gnu
  - attrs=21.4.0=pyhd3eb1b0_0
  - ca-certificates=2025.2.25=h06a4308_0
  - certifi=2021.5.30=py36h06a4308_0
  - importlib-metadata=4.8.1=py36h06a4308_0
  - importlib_metadata=4.8.1=hd3eb1b0_0
  - iniconfig=1.1.1=pyhd3eb1b0_0
  - ld_impl_linux-64=2.40=h12ee557_0
  - libffi=3.3=he6710b0_2
  - libgcc-ng=11.2.0=h1234567_1
  - libgomp=11.2.0=h1234567_1
  - libstdcxx-ng=11.2.0=h1234567_1
  - more-itertools=8.12.0=pyhd3eb1b0_0
  - ncurses=6.4=h6a678d5_0
  - openssl=1.1.1w=h7f8727e_0
  - packaging=21.3=pyhd3eb1b0_0
  - pip=21.2.2=py36h06a4308_0
  - pluggy=0.13.1=py36h06a4308_0
  - py=1.11.0=pyhd3eb1b0_0
  - pyparsing=3.0.4=pyhd3eb1b0_0
  - pytest=6.2.4=py36h06a4308_2
  - python=3.6.13=h12debd9_1
  - readline=8.2=h5eee18b_0
  - setuptools=58.0.4=py36h06a4308_0
  - sqlite=3.45.3=h5eee18b_0
  - tk=8.6.14=h39e8969_0
  - toml=0.10.2=pyhd3eb1b0_0
  - typing_extensions=4.1.1=pyh06a4308_0
  - wheel=0.37.1=pyhd3eb1b0_0
  - xz=5.6.4=h5eee18b_1
  - zipp=3.6.0=pyhd3eb1b0_0
  - zlib=1.2.13=h5eee18b_1
  - pip:
      - coverage==6.2
      - cramjam==2.5.0
      - lz4tools==1.3.1.2
      - mock==5.2.0
      - pytest-catchlog==1.2.2
      - pytest-cov==4.0.0
      - pytest-mock==3.6.1
      - pytest-sugar==0.9.6
      - python-snappy==0.7.3
      - six==1.17.0
      - termcolor==1.1.0
      - tomli==1.2.3
      - xxhash==3.2.0
prefix: /opt/conda/envs/kafka-python

test/test_coordinator.py::test_send_offset_commit_request_versions[api_version0-OffsetCommitRequest_v0] test/test_coordinator.py::test_send_offset_fetch_request_versions[api_version0-OffsetFetchRequest_v0]
test/test_coordinator.py::test_init test/test_coordinator.py::test_autocommit_enable_api_version[api_version0] test/test_coordinator.py::test_autocommit_enable_api_version[api_version1] test/test_coordinator.py::test_autocommit_enable_api_version[api_version2] test/test_coordinator.py::test_autocommit_enable_api_version[api_version3] test/test_coordinator.py::test_protocol_type test/test_coordinator.py::test_group_protocols test/test_coordinator.py::test_pattern_subscription[api_version0] test/test_coordinator.py::test_pattern_subscription[api_version1] test/test_coordinator.py::test_pattern_subscription[api_version2] test/test_coordinator.py::test_pattern_subscription[api_version3] test/test_coordinator.py::test_lookup_assignor test/test_coordinator.py::test_join_complete test/test_coordinator.py::test_subscription_listener test/test_coordinator.py::test_subscription_listener_failure test/test_coordinator.py::test_perform_assignment test/test_coordinator.py::test_on_join_prepare test/test_coordinator.py::test_need_rejoin test/test_coordinator.py::test_refresh_committed_offsets_if_needed test/test_coordinator.py::test_fetch_committed_offsets test/test_coordinator.py::test_close test/test_coordinator.py::test_commit_offsets_async test/test_coordinator.py::test_commit_offsets_sync test/test_coordinator.py::test_maybe_auto_commit_offsets_sync[api_version0-foobar-True-None-False-False-True-False] test/test_coordinator.py::test_maybe_auto_commit_offsets_sync[api_version1-foobar-False-None-False-False-False-False] test/test_coordinator.py::test_maybe_auto_commit_offsets_sync[api_version2-foobar-True-error2-True-True-True-False] test/test_coordinator.py::test_maybe_auto_commit_offsets_sync[api_version3-foobar-True-error3-True-True-True-False] test/test_coordinator.py::test_maybe_auto_commit_offsets_sync[api_version4-foobar-True-error4-True-True-True-False] test/test_coordinator.py::test_maybe_auto_commit_offsets_sync[api_version5-foobar-True-error5-True-True-False-True] test/test_coordinator.py::test_maybe_auto_commit_offsets_sync[api_version6-foobar-True-None-True-True-False-False] test/test_coordinator.py::test_maybe_auto_commit_offsets_sync[api_version7-None-True-None-False-False-True-False] test/test_coordinator.py::test_send_offset_commit_request_fail test/test_coordinator.py::test_send_offset_commit_request_versions[api_version1-OffsetCommitRequest_v1] test/test_coordinator.py::test_send_offset_commit_request_versions[api_version2-OffsetCommitRequest_v2] test/test_coordinator.py::test_send_offset_commit_request_failure test/test_coordinator.py::test_send_offset_commit_request_success test/test_coordinator.py::test_handle_offset_commit_response[response0-GroupAuthorizationFailedError-False-False] test/test_coordinator.py::test_handle_offset_commit_response[response1-OffsetMetadataTooLargeError-False-False] test/test_coordinator.py::test_handle_offset_commit_response[response2-InvalidCommitOffsetSizeError-False-False] test/test_coordinator.py::test_handle_offset_commit_response[response3-GroupLoadInProgressError-False-False] test/test_coordinator.py::test_handle_offset_commit_response[response4-GroupCoordinatorNotAvailableError-True-False] test/test_coordinator.py::test_handle_offset_commit_response[response5-NotCoordinatorForGroupError-True-False] test/test_coordinator.py::test_handle_offset_commit_response[response6-RequestTimedOutError-True-False] test/test_coordinator.py::test_handle_offset_commit_response[response7-CommitFailedError-False-True] test/test_coordinator.py::test_handle_offset_commit_response[response8-CommitFailedError-False-True] test/test_coordinator.py::test_handle_offset_commit_response[response9-CommitFailedError-False-True] test/test_coordinator.py::test_handle_offset_commit_response[response10-InvalidTopicError-False-False] test/test_coordinator.py::test_handle_offset_commit_response[response11-TopicAuthorizationFailedError-False-False] test/test_coordinator.py::test_send_offset_fetch_request_fail test/test_coordinator.py::test_send_offset_fetch_request_versions[api_version1-OffsetFetchRequest_v1] test/test_coordinator.py::test_send_offset_fetch_request_versions[api_version2-OffsetFetchRequest_v1] test/test_coordinator.py::test_send_offset_fetch_request_failure test/test_coordinator.py::test_send_offset_fetch_request_success test/test_coordinator.py::test_handle_offset_fetch_response[response0-GroupLoadInProgressError-False-False] test/test_coordinator.py::test_handle_offset_fetch_response[response1-NotCoordinatorForGroupError-True-False] test/test_coordinator.py::test_handle_offset_fetch_response[response2-UnknownMemberIdError-False-True] test/test_coordinator.py::test_handle_offset_fetch_response[response3-IllegalGenerationError-False-True] test/test_coordinator.py::test_handle_offset_fetch_response[response4-TopicAuthorizationFailedError-False-False] test/test_coordinator.py::test_handle_offset_fetch_response[response5-None-False-False] test/test_coordinator.py::test_heartbeat
kafka/coordinator/consumer.py kafka/coordinator/base.py
kafka/coordinator/consumer.py kafka/coordinator/base.py