Контекст и таймауты в kafka-go

Go-программа без дедлайнов и сигналов отмены ждёт бесконечно: она честно висит, пока кто-то снаружи не решит, что пора заканчиваться. В kafka-go этот «кто-то» — контекст. 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 с явными таймаутами

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, чтобы ограничить время ожидания ответа брокера. Что он найдёт на самом деле?

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; здесь же берём 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 здесь превращает завершение в бесконечный спин. Ошибки из горутины чтения уходят в буферизованный канал: буфер на единицу гарантирует, что отправитель не заблокируется, даже если основной код уже занят выходом. Сообщение, полученное, но не обработанное до конца, остаётся незакоммиченным — офсет не двинулся, после рестарта событие перечитается, и об этом можно прочитать подробнее в уроке про коммит офсетов. Финал остановки — reader.Close(): он разблокирует FetchMessage через io.EOF, если горутина ещё ждала сообщений.

Цикл чтения с остановкой по SIGTERM

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 ещё не закончил. Что произойдёт с этим сообщением после завершения процесса?

Что вернёт FetchMessage при истёкшем дедлайне?

Прочитайте код и выберите, что он выведет:

ctx, cancel := context.WithTimeout(
        context.Background(), 3*time.Second)
defer cancel()

// брокер молчит, сообщений в партиции нет
m, err := reader.FetchMessage(ctx)
log.Println(m, err)

Таймаут против ретрая: судьба события при отмене#

Отмена контекста — это команда «остановись поскорее», а не «отмени последствия». Посреди 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, имя метода замера отставания и конструктор контекста с дедлайном.

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)

Соберите цикл чтения с остановкой

Расставьте строки: Reader, контекст с сигналами, чтение, выход по ошибке, обработка, коммит, финальное закрытие читателя.

Строки

Ваша программа

Нажимайте на строки слева, чтобы собрать программу

Карточки: контекст, таймауты, lag

Термин

Dialer.Timeout

нажмите, чтобы перевернуть

Определение

Максимум времени на установку TCP-соединения; ноль — без таймаута, DefaultDialer ставит десять секунд.

1 / 5

Контекст в kafka-go — не формальность сигнатур, а несущая конструкция жизненного цикла: он прерывает FetchMessage на середине ожидания, догоняет коммиты в группе и обрывает ретраи Writer'а, при этом никогда не откатывает то, что уже произошло. Таймауты лежат в трёх слоях: Dialer.Timeout отвечает за голый dial, Transport — за DialTimeout с рукопожатиями, IdleTimeout и MetadataTTL за соединения и метаданные, а живые запросы ограничиваются дедлайнами соединения и вашим контекстом. ReadLag даёт быстрый замер отставания, пока читатель без группы, и честно отказывается работать там, где офсетами владеет брокер — за это он и ценен: ошибка приходит немедленно и с внятным именем, а не через час тихого зависания в проде. Дальше по курсу: урок про батчи покажет, как внешний ctx взаимодействует с собственными ретраями Writer'а; урок про коммит офсетов разложит по полочкам перечитывание после отмены; а урок про мониторинг lag вырастит замер из этой страницы в полноценный дашборд группы. Начните в песочнице с имитации сигнала, затем повторите эксперимент локально — и разница между отменой и закрытием Reader станет рефлексом. Экономия здесь прямая: один грамотный ctx вместо трёх вечерних дебагов зависшей горутины.

Песочница урока

Редактируйте код и запускайте его прямо здесь — интерпретатор работает в браузере, ничего устанавливать не нужно. За каждый запуск — +2 XP.

Вывод

Запустите код, чтобы увидеть вывод.

Коротко: что запомнить

Контекст — единственный канал остановки 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 удобен: он сам откроет соединение к лидеру и вернёт разницу офсетов, а дедлайн возьмёт из вашего контекста.

Термины урока

Ключевые понятия из этого урока — короткие определения в глоссарии. Наведите на термин, чтобы вспомнить, откройте карточку для подробностей и ссылки на практику.