| 17 | } |
| 18 | |
| 19 | func setupTopic(addr string) { |
| 20 | topics := []string{"trace", "test", "example"} |
| 21 | |
| 22 | conn, err := kafka.Dial("tcp", addr) |
| 23 | if err != nil { |
| 24 | panic(err.Error()) |
| 25 | } |
| 26 | defer conn.Close() |
| 27 | |
| 28 | controller, err := conn.Controller() |
| 29 | if err != nil { |
| 30 | panic(err.Error()) |
| 31 | } |
| 32 | var controllerConn *kafka.Conn |
| 33 | controllerConn, err = kafka.Dial("tcp", net.JoinHostPort(controller.Host, strconv.Itoa(controller.Port))) |
| 34 | if err != nil { |
| 35 | panic(err.Error()) |
| 36 | } |
| 37 | defer controllerConn.Close() |
| 38 | |
| 39 | topicConfigs := make([]kafka.TopicConfig, 0) |
| 40 | for _, topic := range topics { |
| 41 | topicConfigs = append(topicConfigs, kafka.TopicConfig{ |
| 42 | Topic: topic, |
| 43 | NumPartitions: 1, |
| 44 | ReplicationFactor: 1, |
| 45 | }) |
| 46 | } |
| 47 | |
| 48 | err = controllerConn.CreateTopics(topicConfigs...) |
| 49 | if err != nil { |
| 50 | panic(err.Error()) |
| 51 | } |
| 52 | } |