Function testReaderTopicRecreated

reader_test.go:1979–1998

The current behavior of the Reader is to retry OffsetOutOfRange errors indefinitely, which results in programs hanging in the event of a topic being re-created while a consumer is running. To retain backwards-compatibility, ReaderConfig.OffsetOutOfRangeError is being used to instruct the Reader to r

(t *testing.T, ctx context.Context, r *Reader)

// ReaderConfig.OffsetOutOfRangeError is being used to instruct the Reader to
// return an error in this case instead, allowing callers to react.
func testReaderTopicRecreated(t *testing.T, ctx context.Context, r *Reader) {
	r.config.OffsetOutOfRangeError = true

	topic := r.config.Topic

	// add 1 message to the topic
	prepareReader(t, ctx, r, makeTestSequence(1)...)

	// consume the message (moving the offset from 0 -> 1)
	_, err := r.ReadMessage(ctx)
	require.NoError(t, err)

	// destroy the topic, then recreate it so the offset now becomes 0
	deleteTopic(t, topic)
	createTopic(t, topic, 1)

	// expect an error, since the offset should now be out of range
	_, err = r.ReadMessage(ctx)
	require.ErrorIs(t, err, OffsetOutOfRange)
}