--- title: "Контекст и таймауты в kafka-go" description: "DialTimeout, ReadLag, контекст отмены чтения: таймауты Dialer и Transport, замер отставания консьюмера и graceful shutdown по SIGTERM без потери событий." course: "Kafka в Go: продюсеры и консьюмеры на kafka-go" url: "https://golangmaster.ru/courses/kafka-go/lessons/18-kontekst-tajmauty/" published: "2026-09-19" updated: "2026-09-29" minutes: 18 --- # Контекст и таймауты в kafka-go *Курс «Kafka в Go: продюсеры и консьюмеры на kafka-go», урок 18 из 25. ~18 минут. Обновлено 29.09.2026. Источник: https://golangmaster.ru/courses/kafka-go/lessons/18-kontekst-tajmauty/* Контекст — единственный способ сказать kafka-go «пора останавливаться»: каждый метод клиента — WriteMessages, ReadMessage, FetchMessage, CommitMessages, ReadLag — принимает ctx первым аргументом и превращает отмену в ошибку вызова. Разберём, где в клиенте живут таймауты соединений, как измерить отставание консьюмера через ReadLag и как собрать цикл чтения с graceful shutdown, при котором SIGTERM останавливает работу аккуратно, а не рвёт её посреди коммита. Go-программа без дедлайнов и сигналов отмены ждёт бесконечно: она честно висит, пока кто-то снаружи не решит, что пора заканчиваться. В kafka-go этот «кто-то» — контекст. [Reader из урока про API чтения](/courses/kafka-go/lessons/14-reader-api/) блокирующий по устройству: FetchMessage висит до появления сообщения, WriteMessages — до исхода записи, CommitMessages — до ответа брокера. Каждая операция может занять миллисекунды, а может — пока не кончится сеть. Контекст проходит через все вызовы первым аргументом и несёт два сигнала: отмену через cancel() при завершении процесса и дедлайн через WithTimeout или WithDeadline. Когда сигнал срабатывает, блокирующий вызов просыпается и возвращает управление с ошибкой ctx.Err() — context.Canceled или context.DeadlineExceeded. Это не украшение интерфейса, а единственный канал, по которому владелец процесса сообщает библиотеке, что мир снаружи изменился: деплой, масштабирование, смерть пода. Без контекста горутина с Reader переживает всё — она висит на заборе, пока оркестратор не прибьёт процесс целиком вместе с недоставленными сообщениями. ### Один ctx на весь путь события Контекст в kafka-go сквозной: один и тот же ctx сопровождает событие от чтения до коммита. ReadMessage(ctx) внутри устроен как связка FetchMessage(ctx) плюс CommitMessages(ctx) — то есть и чтение, и коммит прерываются одной отменой; в группе это заодно означает, что автокоммит из шестого урока не сработает, если отменить ctx между двумя шагами. У Writer та же картина: WriteMessages(ctx, msgs...) при отмене возвращает ctx.Err(), и все ещё не подтверждённые сообщения пачки из десятого урока теряют гарантии этой попытки — читатель уже принял решение их не ждать. Внутри библиотеки отмена реализована скучно и правильно: каждый блокирующий участок сидит в select на ctx.Done(), и срабатывание канала возвращает ctx.Err() наружу. Практический вывод: не выдумывайте отдельные таймауты на каждый вызов, пока не измерили реальную нагрузку — сначала один общий дедлайн на цикл чтения, затем точечные WithTimeout только там, где профилирование показало затык. Контекст дешевле конфигурации: он меняется на ходу, без пересборки Reader и Writer. ### Таймауты соединения: Dialer и Transport Таймауты установки соединения живут в двух местах, и важно не путать их имена. Dialer из dialer.go отвечает за низкоуровневый dial: поле Timeout — максимум времени на установку TCP-соединения, ноль означает отсутствие таймаута; пакетный DefaultDialer, который используется, когда Dialer не задан, ставит Timeout в десять секунд и включает DualStack. Transport — транспорт для Writer и Reader — несёт собственные ручки: DialTimeout (по умолчанию пять секунд) ограничивает соединение вместе со всеми рукопожатиями — TLS и SASL-переговорами; IdleTimeout (тридцать секунд) закрывает простаивающие соединения; MetadataTTL (шесть секунд) задаёт срок жизни кэша метаданных. Отдельного поля ReadTimeout или WriteTimeout у Transport нет: времена ожидания отдельных запросов контролируются дедлайнами соединения — методами Conn.SetReadDeadline, SetWriteDeadline и SetDeadline, и сама библиотека пользуется этим внутри, например ReadLag ставит SetDeadline из дедлайна контекста. Писать конфиг с несуществующими полями бесполезно — компилятор честно отвергнет, поэтому имена стоит сверять с исходниками своей версии. ### Dialer и Transport с явными таймаутами ```go dialer := &kafka.Dialer{ Timeout: 5 * time.Second, // установка TCP-соединения KeepAlive: 30 * time.Second, } transport := &kafka.Transport{ DialTimeout: 5 * time.Second, // dial плюс TLS и SASL IdleTimeout: time.Minute, // закрытие простаивающих соединений MetadataTTL: 10 * time.Second, // срок жизни кэша метаданных } reader := kafka.NewReader(kafka.ReaderConfig{ Brokers: []string{"broker-1:9092"}, Topic: "payments", GroupID: "payments-workers", Dialer: dialer, Transport: transport, }) ``` Таймауты расставлены по уровням: Dialer.Timeout ограничивает голый dial, Transport.DialTimeout — соединение вместе с рукопожатиями TLS и SASL, поэтому второй перекрывает первый по смыслу. IdleTimeout и MetadataTTL защищают долгоживущий Reader от накопления мёртвых соединений и устаревшей картины партиций. Поле Transport в ReaderConfig принимает RoundTripper, и указатель на kafka.Transport ему подходит. ### Проверка: таймауты Transport **Вопрос:** Разработчик ищет в kafka.Transport поле ReadTimeout, чтобы ограничить время ожидания ответа брокера. Что он найдёт на самом деле? - ✅ Такого поля нет: отдельные запросы ограничиваются дедлайнами соединения через Conn.SetReadDeadline - Поле ReadTimeout есть, но по умолчанию выключено нулевым значением - Вместо него работает WriteTimeout: чтение всегда наследует дедлайн записи - Поле называется RequestTimeout и живёт в Dialer, а не в Transport У Transport в kafka-go нет полей ReadTimeout и WriteTimeout: он отвечает за установку соединений (DialTimeout), их простой (IdleTimeout) и кэш метаданных (MetadataTTL). Тонкий контроль ожидания отдельного запроса достигается дедлайнами соединения — SetReadDeadline, SetWriteDeadline, SetDeadline, которые библиотека ставит и из контекста. ### ReadLag: сколько отстаёт консьюмер Отставание консьюмера — это разница между последним офсетом партиции и позицией читателя, и в kafka-go она измеряется вызовом ReadLag. Сигнатура из reader.go: func (r *Reader) ReadLag(ctx context.Context) (lag int64, err error). Метод открывает соединение к лидеру партиции, читает граничные офсеты и сравнивает с текущей позицией; дедлайн соединению ставится прямо из контекста — conn.SetDeadline(deadline, _ := ctx.Deadline()), поэтому без дедлайна в ctx зависший брокер подвесит и замер. Две границы применимости: ReadLag работает с одним Reader, то есть с одной партицией — для группы читателей нужен замер на каждого участника; и он недоступен читателю с GroupID — при попытке вернётся ошибка errNotAvailableWithGroup с текстом «unavailable when GroupID is set», потому что позиция группы живёт не в Reader, а у брокера. Полноценный мониторинг отставания всей группы строится иначе — через запросы к брокеру по смещениям группы, это отдельная тема [урока про мониторинг lag](/courses/kafka-go/lessons/21-monitoring-lag/); здесь же берём ReadLag как быстрый замер для читателей без группы. > **Живое место: ReadLag против группы.** Инцидент: команда добавила ReadLag в health-check читателя, не заметив, что в конфиге стоит GroupID, и каждый probe-запрос падал с ошибкой «unavailable when GroupID is set». Чинить это понижением уровня лога — путь в никуда: метод принципиально не умеет считать lag группы, его поведение зафиксировано в исходниках reader.go. Рабочие варианты: замерять lag отдельным Reader без GroupID на ту же партицию, либо перейти на брокерские запросы офсетов группы из урока про мониторинг. Мораль шире одного метода: у kafka-go каждый метод имеет границы применимости, и они живут не в интуиции, а в первых строках функции — при странном поведении первым делом сверяйтесь с исходником своей версии, а не со статьёй. ### Graceful shutdown: SIGTERM в cancel() Самый частый сценарий контекста — остановка по сигналу. Пакет signal даёт готовую связку: signal.NotifyContext оборачивает сигналы SIGTERM и os.Interrupt в отмену контекста одной строкой, а cancel() вызывается defer'ом для случая, когда процесс завершился сам без сигналов. Дальше дисциплина цикла чтения: FetchMessage возвращает либо сообщение, либо ошибку — io.EOF, когда Reader закрыли, либо ctx.Err() при отмене; каждый случай разбирается явно, молчаливый continue здесь превращает завершение в бесконечный спин. Ошибки из горутины чтения уходят в буферизованный канал: буфер на единицу гарантирует, что отправитель не заблокируется, даже если основной код уже занят выходом. Сообщение, полученное, но не обработанное до конца, остаётся незакоммиченным — офсет не двинулся, после рестарта событие перечитается, и об этом можно прочитать подробнее в [уроке про коммит офсетов](/courses/kafka-go/lessons/08-kommit-ofsetov/). Финал остановки — reader.Close(): он разблокирует FetchMessage через io.EOF, если горутина ещё ждала сообщений. ### Цикл чтения с остановкой по SIGTERM ```go ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, os.Interrupt) defer stop() errch := make(chan error, 1) // буфер: горутина не зависнет без читателя go func() { for { m, err := reader.FetchMessage(ctx) if err != nil { errch <- err // io.EOF или ctx.Err() return } if err := handle(ctx, m); err != nil { errch <- err return } if err := reader.CommitMessages(ctx, m); err != nil { errch <- err return } } }() select { case err := <-errch: log.Printf("reader stopped: %v", err) case <-ctx.Done(): log.Println("SIGTERM: останавливаемся") } reader.Close() ``` NotifyContext превращает SIGTERM и Ctrl-C в отмену ctx, и FetchMessage внутри select на ctx.Done() просыпается с context.Canceled. Буферизованный канал ошибок нужен затем, что после срабатывания select основной код может уже не читать канал, а небуферизованная отправка оставила бы горутину навсегда. reader.Close() после выхода — гарантия, что зависший FetchMessage тоже разблокируется, вернув io.EOF. ### Проверка: судьба недообработанного сообщения **Вопрос:** SIGTERM пришёл в момент, когда FetchMessage уже вернул сообщение, а handle ещё не закончил. Что произойдёт с этим сообщением после завершения процесса? - ✅ Оно останется незакоммиченным: после рестарта группа перечитает его заново - Оно закоммитится автоматически: Reader коммитит при отмене контекста сам - Оно исчезнет навсегда: отмена контекста стирает событие из партиции - Оно уйдёт в retry-топик: shutdown запускает перенос необработанных событий Офсет двигается только коммитом, и на середине обработки он ещё не двигался: перечитывание после рестарта — штатное поведение at-least-once, заодно требующее идемпотентного хендлера. Никакого автокоммита при отмене в kafka-go нет, а retry-топики — конструкция из следующего урока, которую надо построить руками. ### Что вернёт FetchMessage при истёкшем дедлайне? ```go ctx, cancel := context.WithTimeout( context.Background(), 3*time.Second) defer cancel() // брокер молчит, сообщений в партиции нет m, err := reader.FetchMessage(ctx) log.Println(m, err) ``` **Что выведет программа?** - ✅ Нулевой Message и context.DeadlineExceeded: дедлайн прервал блокирующее ожидание - Нулевой Message и io.EOF: Reader считает себя закрытым - Пустые значения без ошибки: тайм-аут читателя включается автоматически - Вызов заблокируется навсегда: FetchMessage игнорирует контекст FetchMessage сидит в select между потоком сообщений и ctx.Done(): истёкший дедлайн закрывает канал контекста, и метод возвращает ctx.Err() — для WithTimeout это context.DeadlineExceeded. io.EOF зарезервирован за закрытием Reader через Close, а автоматических тайм-аутов у блокирующего чтения нет — только ваш контекст. ### Таймаут против ретрая: судьба события при отмене Отмена контекста — это команда «остановись поскорее», а не «отмени последствия». Посреди FetchMessage она означает, что событие могло уже дойти до хендлера и даже обработаться — но коммит за ним не успел: офсет остался, после рестарта придёт дубликат, и код обязан это переваривать идемпотентностью, как в девятом уроке о гарантиях. Посреди WriteMessages отмена ещё коварнее: Writer мог успеть передать пачку на брокер до срабатывания ctx.Done(), а вызывающий код получил context.Canceled и решит перезаписать события — на продюсере появятся дубликаты, потому что отмена не откатывает уже записанное. Ретраи Writer'а из десятого урока тоже не всесильны: библиотека повторяет пачку сама, но внешний ctx способна оборвать и её собственные попытки. Отсюда правило гигиены: дедлайны на запись ставьте щедрее, чем реальная латентность брокера, а при получении ctx.Err() не пытайтесь доделывать работу в том же вызове — либо у вас ещё жив родительский контекст, либо операцию переносит на себя владелец жизненного цикла. > **Про песочницу: контекст вне сети.** Интерактивы этого урока исполняются в песочнице курса на программах в чистом Go без сети и настоящих сигналов: signal.NotifyContext требует живой ОС, поэтому в редакторе урока он имитируется обычной отменой контекста. Локально всё встаёт на место: поднимите брокер в Docker из третьего урока, запустите цикл чтения и отправьте процессу SIGTERM командой kill — увидите в логе context.Canceled от FetchMessage, а после перезапуска — перечитанное незакоммиченное сообщение. Этот двухшаговый эксперимент объясняет at-least-once нагляднее любой диаграммы: одна и та же строка кода ведёт себя по-разному до и после сигнала, и вся разница сидит в одном коммите. ### Правила контекста и таймаутов одним списком - Один ctx проходит весь путь события: ReadMessage(ctx) — это FetchMessage плюс CommitMessages с одним и тем же контекстом. - ctx.Err() различайте по типам: io.EOF — Reader закрыт, context.Canceled — остановка, DeadlineExceeded — дедлайн. - Таймауты соединения: Dialer.Timeout для dial (DefaultDialer — десять секунд), Transport.DialTimeout с TLS и SASL (пять секунд). - Полей ReadTimeout и WriteTimeout у Transport нет: запросы ограничиваются дедлайнами через Conn.SetReadDeadline и SetWriteDeadline. - ReadLag(ctx) — замер одной партиции без группы: с GroupID вернёт errNotAvailableWithGroup. - Сообщению, полученному до отмены ctx, коммит не гарантируется: считайте перечитывание нормой и держите хендлер идемпотентным. ### Карта таймаутов kafka-go | Механизм | За что отвечает | Дефолт | | --- | --- | --- | | Dialer.Timeout | Установка TCP-соединения без рукопожатий | Ноль — без таймаута; DefaultDialer — 10s | | Transport.DialTimeout | Соединение вместе с TLS и SASL-переговорами | 5s | | Transport.IdleTimeout | Жизнь простаивающих соединений | 30s | | Transport.MetadataTTL | Срок кэша метаданных топиков и партиций | 6s | | ctx с WithTimeout | Отмена блокирующих вызовов API: чтение, запись, коммит | Без дедлайна — ожидание вечно | ### Соберите замер lag с дедлайном Допишите сигнал в NotifyContext, имя метода замера отставания и конструктор контекста с дедлайном. ```go ctx, stop := signal.NotifyContext(context.Background(), syscall.___, os.Interrupt) defer stop() rctx, cancel := context.___(ctx, 3*time.Second) defer cancel() lag, err := reader.___(rctx) if err != nil { log.Fatalf("lag: %v", err) } log.Printf("отставание: %d сообщений", lag) ``` Подсказка: Сигнал завершения называется SIGTERM, дедлайн строит context.WithTimeout, а отставание измеряет ReadLag — он вернёт lag и ошибку. Правильные ответы в пропусках: `SIGTERM`, `WithTimeout`, `ReadLag` ### Соберите цикл чтения с остановкой Расставьте строки: Reader, контекст с сигналами, чтение, выход по ошибке, обработка, коммит, финальное закрытие читателя. Правильный порядок строк: 1. `reader := kafka.NewReader(kafka.ReaderConfig{Topic: "events", GroupID: "workers"})` 2. `ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, os.Interrupt)` 3. `defer stop()` 4. `m, err := reader.FetchMessage(ctx)` 5. `if err != nil { return }` 6. `if err := handle(ctx, m); err != nil { return }` 7. `if err := reader.CommitMessages(ctx, m); err != nil { return }` 8. `reader.Close()` ### Карточки: контекст, таймауты, lag - **Dialer.Timeout** — Максимум времени на установку TCP-соединения; ноль — без таймаута, DefaultDialer ставит десять секунд. - **Transport.DialTimeout** — Дедлайн соединения со всеми рукопожатиями TLS и SASL; дефолт пять секунд. IdleTimeout и MetadataTTL — рядом. - **FetchMessage(ctx)** — Блокирующее чтение без автокоммита: отмена ctx возвращает ctx.Err(), закрытый Reader — io.EOF. - **ReadLag(ctx)** — Разница между последним офсетом партиции и позицией читателя; при GroupID недоступен — errNotAvailableWithGroup. - **context.DeadlineExceeded** — Ошибка истёкшего дедлайна: вызов прерван, но сделанные ранее последствия — записи, обработка — не откатываются. Контекст в kafka-go — не формальность сигнатур, а несущая конструкция жизненного цикла: он прерывает FetchMessage на середине ожидания, догоняет коммиты в группе и обрывает ретраи Writer'а, при этом никогда не откатывает то, что уже произошло. Таймауты лежат в трёх слоях: Dialer.Timeout отвечает за голый dial, Transport — за DialTimeout с рукопожатиями, IdleTimeout и MetadataTTL за соединения и метаданные, а живые запросы ограничиваются дедлайнами соединения и вашим контекстом. ReadLag даёт быстрый замер отставания, пока читатель без группы, и честно отказывается работать там, где офсетами владеет брокер — за это он и ценен: ошибка приходит немедленно и с внятным именем, а не через час тихого зависания в проде. Дальше по курсу: [урок про батчи](/courses/kafka-go/lessons/10-producer-batchi/) покажет, как внешний ctx взаимодействует с собственными ретраями Writer'а; [урок про коммит офсетов](/courses/kafka-go/lessons/08-kommit-ofsetov/) разложит по полочкам перечитывание после отмены; а [урок про мониторинг lag](/courses/kafka-go/lessons/21-monitoring-lag/) вырастит замер из этой страницы в полноценный дашборд группы. Начните в песочнице с имитации сигнала, затем повторите эксперимент локально — и разница между отменой и закрытием Reader станет рефлексом. Экономия здесь прямая: один грамотный ctx вместо трёх вечерних дебагов зависшей горутины. ## Коротко: что запомнить - Контекст — единственный канал остановки kafka-go: каждый метод клиента принимает ctx и возвращает ctx.Err() при отмене. - FetchMessage(ctx) прерывается отменой или дедлайном; io.EOF означает закрытый Reader — три исхода различаются по типу ошибки. - Таймауты соединений: Dialer.Timeout для dial, Transport.DialTimeout (5s) с TLS и SASL, IdleTimeout (30s), MetadataTTL (6s). - Полей ReadTimeout и WriteTimeout у Transport нет: запросы ограничивают дедлайны Conn.SetReadDeadline и SetWriteDeadline. - ReadLag(ctx) измеряет отставание одного Reader и отказывается работать с GroupID через errNotAvailableWithGroup. - Graceful shutdown: signal.NotifyContext превращает SIGTERM в cancel(), буферизованный канал ошибок не даёт горутине утечь. - Отмена не откатывает последствия: недообработанное сообщение перечитается, записанная пачка останется записанной. ## Домашнее задание Поднимите локальный брокер в Docker из третьего урока и соберите цикл чтения из этого урока: signal.NotifyContext с SIGTERM, буферизованный errch, FetchMessage, handle, CommitMessages. Запустите, отправьте несколько сообщений в топик и пошлите процессу SIGTERM посреди обработки — убедитесь в логах, что FetchMessage вернул context.Canceled, а после перезапуска событие перечиталось. Добавьте замер ReadLag до и после паузы в обработке: сначала на Reader без GroupID, затем с GroupID — зафиксируйте текст ошибки errNotAvailableWithGroup и объясните, почему она честная. Установите Transport с DialTimeout в одну миллисекунду на несуществующий адрес и опишите разницу между этой ошибкой и context.DeadlineExceeded от FetchMessage: где произошёл тайм-аут, кто его отправил и что он говорит о брокере. ## Вопросы и ответы по уроку ### Какой контекст передавать FetchMessage: с дедлайном или без? Для основного цикла чтения — родительский без дедлайна, привязанный к сигналам процесса: сообщения должны ждать, сколько нужно, а остановка придёт через cancel(). Точечные дедлайны уместны на вспомогательных операциях — замер ReadLag, запись в DLQ, обращение к Admin API. Правило простое: дедлайн должен отражать бизнес-срок операции, а не страх перед бесконечностью. Если боитесь зависания, измерьте реальную латентность и поставьте дедлайн с запасом — фальшивые срабатывания хуже отсутствия тайм-аута. ### Чем context.Canceled отличается от io.EOF в цикле чтения? Canceled значит, что владелец контекста приказал остановиться — пришёл SIGTERM или сработал cancel(): это управляемое завершение, при котором недообработанное событие останется незакоммиченным. io.EOF значит, что Reader закрыт вызовом Close(): чтение через него более невозможно, и продолжать цикл бессмысленно даже с живым контекстом. Обе ошибки — нормальные исходы остановки, но сценарии разные: после Canceled процесс может перезапуститься и продолжить с того же офсета, после EOF Reader пересоздают. Молчаливо глотать любую из них — способ получить спин-цикл на сотню процентов CPU. ### ReadLag возвращает отставание всей группы? Нет: метод считает lag одного Reader — одной партиции, и при GroupID в конфиге сразу возвращает errNotAvailableWithGroup с текстом «unavailable when GroupID is set». Позиция группы хранится у брокера, а не в Reader, поэтому честный замер группы делается запросами офсетов группы ко всем партициям топика — в kafka-go это отдельные запросы через Client, а не ReadLag. Для быстрой диагностики одиночного читателя без группы ReadLag удобен: он сам откроет соединение к лидеру и вернёт разницу офсетов, а дедлайн возьмёт из вашего контекста.