diff --git a/kafka/coordinator/consumer.py b/kafka/coordinator/consumer.py
index a5e3067..b2ef1ea 100644
--- a/kafka/coordinator/consumer.py
+++ b/kafka/coordinator/consumer.py
@@ -91,8 +91,10 @@ class ConsumerCoordinator(BaseCoordinator):
                 log.warning('Broker version (%s) does not support offset'
                             ' commits; disabling auto-commit.',
                             self.config['api_version'])
+                self.config['enable_auto_commit'] = False
             elif self.config['group_id'] is None:
                 log.warning('group_id is None: disabling auto-commit.')
+                self.config['enable_auto_commit'] = False
             else:
                 interval = self.config['auto_commit_interval_ms'] / 1000.0
                 self._auto_commit_task = AutoCommitTask(weakref.proxy(self), interval)
@@ -192,7 +194,7 @@ class ConsumerCoordinator(BaseCoordinator):
         assignor.on_assignment(assignment)
 
         # restart the autocommit task if needed
-        if self.config['enable_auto_commit']:
+        if self._auto_commit_task:
             self._auto_commit_task.enable()
 
         assigned = set(self._subscription.assigned_partitions())
@@ -364,27 +366,27 @@ class ConsumerCoordinator(BaseCoordinator):
             time.sleep(self.config['retry_backoff_ms'] / 1000.0)
 
     def _maybe_auto_commit_offsets_sync(self):
-        if self.config['api_version'] < (0, 8, 1):
+        if self._auto_commit_task is None:
             return
-        if self.config['enable_auto_commit']:
-            # disable periodic commits prior to committing synchronously. note that they will
-            # be re-enabled after a rebalance completes
-            self._auto_commit_task.disable()
 
-            try:
-                self.commit_offsets_sync(self._subscription.all_consumed_offsets())
-
-            # The three main group membership errors are known and should not
-            # require a stacktrace -- just a warning
-            except (Errors.UnknownMemberIdError,
-                    Errors.IllegalGenerationError,
-                    Errors.RebalanceInProgressError):
-                log.warning("Offset commit failed: group membership out of date"
-                            " This is likely to cause duplicate message"
-                            " delivery.")
-            except Exception:
-                log.exception("Offset commit failed: This is likely to cause"
-                              " duplicate message delivery")
+        # disable periodic commits prior to committing synchronously. note that they will
+        # be re-enabled after a rebalance completes
+        self._auto_commit_task.disable()
+
+        try:
+            self.commit_offsets_sync(self._subscription.all_consumed_offsets())
+
+        # The three main group membership errors are known and should not
+        # require a stacktrace -- just a warning
+        except (Errors.UnknownMemberIdError,
+                Errors.IllegalGenerationError,
+                Errors.RebalanceInProgressError):
+            log.warning("Offset commit failed: group membership out of date"
+                        " This is likely to cause duplicate message"
+                        " delivery.")
+        except Exception:
+            log.exception("Offset commit failed: This is likely to cause"
+                          " duplicate message delivery")
 
     def _send_offset_commit_request(self, offsets):
         """Commit offsets for the specified list of topics and partitions.

Consumer exception on close when group id is None
Following the conversation in #601, setting the `group_id` to `None` in a Consumer causes an exception to be raised when the consumer is closed.

```
>>> from kafka import KafkaConsumer
>>> k = KafkaConsumer('example', bootstrap_servers=['server'], group_id=None)
>>> k.close()
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/Users/madisonb/.local/share/virtualenvs/scrapy-cluster/lib/python2.7/site-packages/kafka/consumer/group.py", line 257, in close
    self._coordinator.close()
  File "/Users/madisonb/.local/share/virtualenvs/scrapy-cluster/lib/python2.7/site-packages/kafka/coordinator/consumer.py", line 306, in close
    self._maybe_auto_commit_offsets_sync()
  File "/Users/madisonb/.local/share/virtualenvs/scrapy-cluster/lib/python2.7/site-packages/kafka/coordinator/consumer.py", line 372, in _maybe_auto_commit_offsets_sync
    self._auto_commit_task.disable()
AttributeError: 'NoneType' object has no attribute 'disable'
>>> k = KafkaConsumer('example', bootstrap_servers=['server'], group_id='stuff')
>>> k.close()
>>>
```
diff --git a/test/test_coordinator.py b/test/test_coordinator.py
index 847cbc1..44db808 100644
--- a/test/test_coordinator.py
+++ b/test/test_coordinator.py
@@ -52,12 +52,16 @@ def test_init(conn):
 
 @pytest.mark.parametrize("api_version", [(0, 8, 0), (0, 8, 1), (0, 8, 2), (0, 9)])
 def test_autocommit_enable_api_version(conn, api_version):
-    coordinator = ConsumerCoordinator(
-        KafkaClient(), SubscriptionState(), api_version=api_version)
+    coordinator = ConsumerCoordinator(KafkaClient(), SubscriptionState(),
+                                      enable_auto_commit=True,
+                                      group_id='foobar',
+                                      api_version=api_version)
     if api_version < (0, 8, 1):
         assert coordinator._auto_commit_task is None
+        assert coordinator.config['enable_auto_commit'] is False
     else:
         assert coordinator._auto_commit_task is not None
+        assert coordinator.config['enable_auto_commit'] is True
 
 
 def test_protocol_type(coordinator):
@@ -349,28 +353,40 @@ def test_commit_offsets_sync(mocker, coordinator, offsets):
 
 
 @pytest.mark.parametrize(
-    'api_version,enable,error,task_disable,commit_offsets,warn,exc', [
-        ((0, 8), True, None, False, False, False, False),
-        ((0, 9), False, None, False, False, False, False),
-        ((0, 9), True, Errors.UnknownMemberIdError(), True, True, True, False),
-        ((0, 9), True, Errors.IllegalGenerationError(), True, True, True, False),
-        ((0, 9), True, Errors.RebalanceInProgressError(), True, True, True, False),
-        ((0, 9), True, Exception(), True, True, False, True),
-        ((0, 9), True, None, True, True, False, False),
+    'api_version,group_id,enable,error,has_auto_commit,commit_offsets,warn,exc', [
+        ((0, 8), 'foobar', True, None, False, False, True, False),
+        ((0, 9), 'foobar', False, None, False, False, False, False),
+        ((0, 9), 'foobar', True, Errors.UnknownMemberIdError(), True, True, True, False),
+        ((0, 9), 'foobar', True, Errors.IllegalGenerationError(), True, True, True, False),
+        ((0, 9), 'foobar', True, Errors.RebalanceInProgressError(), True, True, True, False),
+        ((0, 9), 'foobar', True, Exception(), True, True, False, True),
+        ((0, 9), 'foobar', True, None, True, True, False, False),
+        ((0, 9), None, True, None, False, False, True, False),
     ])
-def test_maybe_auto_commit_offsets_sync(mocker, coordinator,
-                                        api_version, enable, error, task_disable,
-                                        commit_offsets, warn, exc):
-    auto_commit_task = mocker.patch.object(coordinator, '_auto_commit_task')
-    commit_sync = mocker.patch.object(coordinator, 'commit_offsets_sync',
-                                      side_effect=error)
+def test_maybe_auto_commit_offsets_sync(mocker, api_version, group_id, enable,
+                                        error, has_auto_commit, commit_offsets,
+                                        warn, exc):
     mock_warn = mocker.patch('kafka.coordinator.consumer.log.warning')
     mock_exc = mocker.patch('kafka.coordinator.consumer.log.exception')
+    coordinator = ConsumerCoordinator(KafkaClient(), SubscriptionState(),
+                                      api_version=api_version,
+                                      enable_auto_commit=enable,
+                                      group_id=group_id)
+    commit_sync = mocker.patch.object(coordinator, 'commit_offsets_sync',
+                                      side_effect=error)
+    if has_auto_commit:
+        assert coordinator._auto_commit_task is not None
+        coordinator._auto_commit_task.enable()
+        assert coordinator._auto_commit_task._enabled is True
+    else:
+        assert coordinator._auto_commit_task is None
 
-    coordinator.config['api_version'] = api_version
-    coordinator.config['enable_auto_commit'] = enable
     assert coordinator._maybe_auto_commit_offsets_sync() is None
-    assert auto_commit_task.disable.call_count == (1 if task_disable else 0)
+
+    if has_auto_commit:
+        assert coordinator._auto_commit_task is not None
+        assert coordinator._auto_commit_task._enabled is False
+
     assert commit_sync.call_count == (1 if commit_offsets else 0)
     assert mock_warn.call_count == (1 if warn else 0)
     assert mock_exc.call_count == (1 if exc else 0)

pip install -e .
parse_log_pytest
requirements.txt
pytest pytest-cov pytest-catchlog pytest-pylint pytest-sugar pytest-mock mock python-snappy lz4tools xxhash
apt-get update apt-get install -y libsnappy-dev
3.5
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
astroid==2.11.7
attrs==22.2.0
certifi==2021.5.30
coverage==6.2
cramjam==2.5.0
dill==0.3.4
importlib-metadata==4.8.3
iniconfig==1.1.1
isort==5.10.1
-e git+https://github.com/dpkp/kafka-python.git@b96f4ccf070109a022deb98b569e61d23e4e75b9#egg=kafka_python
lazy-object-proxy==1.7.1
lz4tools==1.3.1.2
mccabe==0.7.0
mock==5.2.0
packaging==21.3
platformdirs==2.4.0
pluggy==1.0.0
py==1.11.0
pylint==2.13.9
pyparsing==3.1.4
pytest==7.0.1
pytest-catchlog==1.2.2
pytest-cov==4.0.0
pytest-mock==3.6.1
pytest-pylint==0.18.0
pytest-sugar==0.9.6
python-snappy==0.7.3
six==1.17.0
termcolor==1.1.0
toml==0.10.2
tomli==1.2.3
typed-ast==1.5.5
typing_extensions==4.1.1
wrapt==1.16.0
xxhash==3.2.0
zipp==3.6.0

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
  - ca-certificates=2025.2.25=h06a4308_0
  - certifi=2021.5.30=py36h06a4308_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
  - ncurses=6.4=h6a678d5_0
  - openssl=1.1.1w=h7f8727e_0
  - pip=21.2.2=py36h06a4308_0
  - 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
  - wheel=0.37.1=pyhd3eb1b0_0
  - xz=5.6.4=h5eee18b_1
  - zlib=1.2.13=h5eee18b_1
  - pip:
      - astroid==2.11.7
      - attrs==22.2.0
      - coverage==6.2
      - cramjam==2.5.0
      - dill==0.3.4
      - importlib-metadata==4.8.3
      - iniconfig==1.1.1
      - isort==5.10.1
      - lazy-object-proxy==1.7.1
      - lz4tools==1.3.1.2
      - mccabe==0.7.0
      - mock==5.2.0
      - packaging==21.3
      - platformdirs==2.4.0
      - pluggy==1.0.0
      - py==1.11.0
      - pylint==2.13.9
      - pyparsing==3.1.4
      - pytest==7.0.1
      - pytest-catchlog==1.2.2
      - pytest-cov==4.0.0
      - pytest-mock==3.6.1
      - pytest-pylint==0.18.0
      - pytest-sugar==0.9.6
      - python-snappy==0.7.3
      - six==1.17.0
      - termcolor==1.1.0
      - toml==0.10.2
      - tomli==1.2.3
      - typed-ast==1.5.5
      - typing-extensions==4.1.1
      - wrapt==1.16.0
      - xxhash==3.2.0
      - zipp==3.6.0
prefix: /opt/conda/envs/kafka-python

test/test_coordinator.py::test_autocommit_enable_api_version[api_version0] test/test_coordinator.py::test_maybe_auto_commit_offsets_sync[api_version7-None-True-None-False-False-True-False]
test/test_coordinator.py::test_handle_offset_commit_response[response10-InvalidTopicError-False-False]
test/test_coordinator.py::test_init 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_send_offset_commit_request_fail test/test_coordinator.py::test_send_offset_commit_request_versions[api_version0-OffsetCommitRequest_v0] 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-UnknownMemberIdError-False-True] test/test_coordinator.py::test_handle_offset_commit_response[response8-IllegalGenerationError-False-True] test/test_coordinator.py::test_handle_offset_commit_response[response9-RebalanceInProgressError-False-True] 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_version0-OffsetFetchRequest_v0] 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/consumer.py