Function TestJetStreamClusterStreamLeaderChangeClientErr

test/js_test.go:9672–9761

(t *testing.T)

}

func TestJetStreamClusterStreamLeaderChangeClientErr(t *testing.T) {
	t.Skip("The 2.9 server changed behavior making this test fail now")

	cfg := &nats.StreamConfig{
		Name:     "TEST",
		Subjects: []string{"foo"},
		Replicas: 3,
	}

	withJSClusterAndStream(t, "R3S", 3, cfg, func(t *testing.T, stream string, servers ...*jsServer) {
		// We want to make sure the worse thing seen by the lower levels during a leadership change is NoResponders.
		// We will have three concurrent contexts going on.
		// 1. Leadership Changes every 500ms.
		// 2. Publishing messages to the stream every 10ms.
		// 3. StreamInfo calls every 15ms.
		expires := time.Now().Add(5 * time.Second)
		var wg sync.WaitGroup
		wg.Add(3)

		randServer := func() *server.Server {
			return servers[mrand.Intn(len(servers))].Server
		}

		// Leadership changes.
		go func() {
			defer wg.Done()
			nc, js := jsClient(t, randServer())
			defer nc.Close()

			sds := fmt.Sprintf(server.JSApiStreamLeaderStepDownT, "TEST")
			for time.Now().Before(expires) {
				time.Sleep(500 * time.Millisecond)
				si, err := js.StreamInfo("TEST")
				expectOk(t, err)
				_, err = nc.Request(sds, nil, time.Second)
				expectOk(t, err)

				// Wait on new leader.
				checkFor(t, 5*time.Second, 50*time.Millisecond, func() error {
					si, err = js.StreamInfo("TEST")
					if err != nil {
						return err
					}
					if si.Cluster.Leader == "" {
						return errors.New("No leader yet")
					}
					return nil
				})
			}
		}()

		// Published every 10ms
		toc := 0
		go func() {
			defer wg.Done()
			nc, js := jsClient(t, randServer())
			defer nc.Close()