Function TestListOffsets

admin_test.go:2188–2301

(t *testing.T)

}

func TestListOffsets(t *testing.T) {
	const topic = "my-topic"

	t.Run("returns offsets from a single broker", func(t *testing.T) {
		const partition = int32(0)
		const timestamp = int64(1690000000000)
		const expectedOffset = int64(42)

		broker := newMockBroker(t, 1)
		broker.SetHandlerByMap(map[string]MockResponse{
			"OffsetRequest":   NewMockOffsetResponse(t).SetOffset(topic, partition, timestamp, expectedOffset),
			"MetadataRequest": mockMetadataFor(t, broker).SetLeader(topic, partition, broker.BrokerID()),
		})

		result, err := newTestAdmin(t, broker).ListOffsets(map[string]map[int32]int64{
			topic: {partition: timestamp},
		}, nil)
		require.NoError(t, err)

		info := result[topic][partition]
		require.NotNil(t, info)
		assert.Equal(t, ErrNoError, info.Err)
		assert.Equal(t, expectedOffset, info.Offset)
	})

	t.Run("fans out across partition leaders", func(t *testing.T) {
		const offset1 = int64(100)
		const offset2 = int64(200)

		leader1 := newMockBroker(t, 1)
		leader2 := newMockBroker(t, 2)
		metadata := mockMetadataFor(t, leader1, leader2).
			SetLeader(topic, 0, leader1.BrokerID()).
			SetLeader(topic, 1, leader2.BrokerID())

		leader1.SetHandlerByMap(map[string]MockResponse{
			"MetadataRequest": metadata,
			"OffsetRequest":   NewMockOffsetResponse(t).SetOffset(topic, 0, OffsetNewest, offset1),
		})
		leader2.SetHandlerByMap(map[string]MockResponse{
			"MetadataRequest": metadata,
			"OffsetRequest":   NewMockOffsetResponse(t).SetOffset(topic, 1, OffsetNewest, offset2),
		})

		result, err := newTestAdmin(t, leader1).ListOffsets(map[string]map[int32]int64{
			topic: {0: OffsetNewest, 1: OffsetNewest},
		}, nil)
		require.NoError(t, err)
		require.Contains(t, result, topic)
		assert.Equal(t, offset1, result[topic][0].Offset)
		assert.Equal(t, offset2, result[topic][1].Offset)
	})

	t.Run("propagates IsolationLevel to the broker request", func(t *testing.T) {
		const partition = int32(0)

		broker := newMockBroker(t, 1)
		metadata := mockMetadataFor(t, broker).SetLeader(topic, partition, broker.BrokerID())