Consumer: handle retriable errors in offsets_for_times (#3022)

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: dpkp

kafka/consumer/fetcher.py

@@ -349,15 +349,22 @@ async def _fetch_offsets_by_times_async(self, timestamps, timeout_ms=None):
                 offsets, retry = await self._manager.wait_for(future, timer.timeout_ms)
             except Errors.KafkaTimeoutError:
                 break
-            except Exception as exc:
-                if not getattr(exc, 'retriable', False):
+            except Errors.KafkaError as exc:
+                if not exc.retriable:
                     raise
-                if getattr(exc, 'invalid_metadata', False) or self._manager.cluster.need_update:
+                if exc.invalid_metadata or self._manager.cluster.need_update:
                     refresh_future = self._manager.cluster.request_update()
                     try:
                         await self._manager.wait_for(refresh_future, timer.timeout_ms)
                     except Errors.KafkaTimeoutError:
                         break
+                    except Errors.KafkaError as refresh_exc:
+                        if not refresh_exc.retriable:
+                            raise
+                        delay = self.config['retry_backoff_ms'] / 1000
+                        if timer.timeout_ms is not None:
+                            delay = min(delay, timer.timeout_ms / 1000)
+                        await self._manager._net.sleep(delay)
                 else:
                     delay = self.config['retry_backoff_ms'] / 1000
                     if timer.timeout_ms is not None: