Function TestIssue806

transport_test.go:170–306

(t *testing.T)

}

func TestIssue806(t *testing.T) {
	// ensure the test times out if the bug is re-introduced
	ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
	defer cancel()

	// simulate unknown topic want auto create with unknownTopicName,
	const unknownTopicName = "unknown-topic"
	const okTopicName = "good-topic"

	// make the connection pool think it's immediately ready to send
	ready := make(chan struct{})
	close(ready)

	// allow the system to wake as much as it wants
	wake := make(chan event)
	defer close(wake)
	go func() {
		for {
			select {
			case <-ctx.Done():
				return
			case e := <-wake:
				if e == nil {
					return
				}
				e.trigger()
			}
		}
	}()

	// handle requests by immediately resolving them with a create topics response,
	// the "unknown topic" will have err UNKNOWN_TOPIC_OR_PARTITION
	requests := make(chan connRequest, 1)
	defer close(requests)
	go func() {
		request := <-requests
		request.res.resolve(&meta.Response{
			Topics: []meta.ResponseTopic{
				{
					Name:      unknownTopicName,
					ErrorCode: int16(UnknownTopicOrPartition),
				},
				{
					Name: okTopicName,
					Partitions: []meta.ResponsePartition{
						{
							PartitionIndex: 0,
						},
					},
				},
			},
		})
	}()

	pool := &connPool{
		ready: ready,
		wake:  wake,
		conns: map[int32]*connGroup{},