Function CreateCluster

pkg/ingest/testkafka/cluster.go:26–48

CreateCluster returns a fake Kafka cluster for unit testing.

(t testing.TB, numPartitions int32, topicName string)

// CreateCluster returns a fake Kafka cluster for unit testing.
func CreateCluster(t testing.TB, numPartitions int32, topicName string) (*Cluster, string) {
	fake, err := kfake.NewCluster(kfake.NumBrokers(1), kfake.SeedTopics(numPartitions, topicName))
	require.NoError(t, err)
	t.Cleanup(fake.Close)

	addrs := fake.ListenAddrs()
	require.Len(t, addrs, 1)

	c := &Cluster{
		t:                t,
		fake:             fake,
		topic:            topicName,
		numPartitions:    int(numPartitions),
		committedOffsets: map[string][]int64{},
		controlFuncs:     map[kmsg.Key]controlFn{},
	}

	// Add support for consumer groups
	c.fake.ControlKey(kmsg.OffsetCommit.Int16(), c.offsetCommit)
	c.fake.ControlKey(kmsg.OffsetFetch.Int16(), c.offsetFetch)

	return c, addrs[0]
}

func (c *Cluster) ControlKey(key kmsg.Key, fn controlFn) {
	switch key {