diff --git a/faust/transport/consumer.py b/faust/transport/consumer.py
index ab3554416..9e49a9a8f 100644
--- a/faust/transport/consumer.py
+++ b/faust/transport/consumer.py
@@ -1077,8 +1077,8 @@ async def _commit_offsets(
                 on_timeout.info("+tables.on_commit")
                 self.app.tables.on_commit(committable_offsets)
                 on_timeout.info("-tables.on_commit")
-        self._committed_offset.update(committable_offsets)
-        self.app.monitor.on_tp_commit(committable_offsets)
+                self._committed_offset.update(committable_offsets)
+                self.app.monitor.on_tp_commit(committable_offsets)
         return did_commit
 
     def _filter_tps_with_pending_acks(
diff --git a/tests/unit/transport/test_consumer.py b/tests/unit/transport/test_consumer.py
index 8a5986e58..d0261179a 100644
--- a/tests/unit/transport/test_consumer.py
+++ b/tests/unit/transport/test_consumer.py
@@ -927,9 +927,10 @@ async def test_handle_attached(self, *, consumer):
 
     @pytest.mark.asyncio
     async def test_commit_offsets(self, *, consumer):
-        consumer._commit = AsyncMock(name="_commit")
+        consumer._commit = AsyncMock(name="_commit", return_value=True)
         consumer.current_assignment.update({TP1, TP2})
         consumer.app.producer.flush = AsyncMock()
+        consumer.app.monitor.on_tp_commit = Mock(name="on_tp_commit")
         await consumer._commit_offsets(
             {
                 TP1: 3003,
@@ -942,6 +943,15 @@ async def test_commit_offsets(self, *, consumer):
                 TP2: 6006,
             }
         )
+        # On a successful commit, bookkeeping must advance.
+        consumer.app.monitor.on_tp_commit.assert_called_once_with(
+            {
+                TP1: 3003,
+                TP2: 6006,
+            }
+        )
+        assert consumer._committed_offset[TP1] == 3003
+        assert consumer._committed_offset[TP2] == 6006
 
     @pytest.mark.asyncio
     async def test_commit_offsets__did_not_commit(self, *, consumer):
@@ -950,6 +960,8 @@ async def test_commit_offsets__did_not_commit(self, *, consumer):
         consumer.app.producer.flush = AsyncMock()
         consumer.current_assignment.update({TP1, TP2})
         consumer.app.tables = Mock(name="app.tables")
+        consumer.app.monitor.on_tp_commit = Mock(name="on_tp_commit")
+        committed_offset_before = dict(consumer._committed_offset)
         await consumer._commit_offsets(
             {
                 TP1: 3003,
@@ -958,6 +970,11 @@ async def test_commit_offsets__did_not_commit(self, *, consumer):
             }
         )
         consumer.app.tables.on_commit.assert_not_called()
+        # Bookkeeping must not advance when the underlying commit failed
+        # (see faust-streaming/faust#316): a failed commit must not make
+        # Faust believe those offsets were actually committed.
+        consumer.app.monitor.on_tp_commit.assert_not_called()
+        assert consumer._committed_offset == committed_offset_before
 
     @pytest.mark.asyncio
     async def test_commit_offsets__in_transaction(self, *, consumer):