Function createTopic

reader_test.go:279–326

(t *testing.T, topic string, partitions int)

}

func createTopic(t *testing.T, topic string, partitions int) {
	t.Helper()

	t.Logf("createTopic(%s, %d)", topic, partitions)

	conn, err := Dial("tcp", "localhost:9092")
	if err != nil {
		err = fmt.Errorf("createTopic, Dial: %w", err)
		t.Fatal(err)
	}
	defer conn.Close()

	controller, err := conn.Controller()
	if err != nil {
		err = fmt.Errorf("createTopic, conn.Controller: %w", err)
		t.Fatal(err)
	}

	conn, err = Dial("tcp", net.JoinHostPort(controller.Host, strconv.Itoa(controller.Port)))
	if err != nil {
		t.Fatal(err)
	}

	conn.SetDeadline(time.Now().Add(10 * time.Second))

	_, err = conn.createTopics(createTopicsRequest{
		Topics: []createTopicsRequestV0Topic{
			{
				Topic:             topic,
				NumPartitions:     int32(partitions),
				ReplicationFactor: 1,
			},
		},
		Timeout: milliseconds(5 * time.Second),
	})
	if err != nil {
		if !errors.Is(err, TopicAlreadyExists) {
			err = fmt.Errorf("createTopic, conn.createTopics: %w", err)
			t.Error(err)
			t.FailNow()
		}
	}

	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
	defer cancel()

	waitForTopic(ctx, t, topic)
}

// Block until topic exists.
func waitForTopic(ctx context.Context, t *testing.T, topic string) {