func TestPipelinedDMLDisableRetry(t *testing.T) {
	// the case tests that
	// 1. auto-retry for pipelined dml is disabled
	// 2. the write conflict error message returned from a Flush (instead of from Commit) is correct
	require.Nil(t, failpoint.Enable("tikvclient/pipelinedMemDBMinFlushKeys", `return(1)`))
	require.Nil(t, failpoint.Enable("tikvclient/pipelinedMemDBMinFlushSize", `return(1)`))
	require.Nil(t, failpoint.Enable("tikvclient/pipelinedMemDBForceFlushSizeThreshold", `return(1)`))
	defer func() {
		require.Nil(t, failpoint.Disable("tikvclient/pipelinedMemDBMinFlushKeys"))
		require.Nil(t, failpoint.Disable("tikvclient/pipelinedMemDBMinFlushSize"))
		require.Nil(t, failpoint.Disable("tikvclient/pipelinedMemDBForceFlushSizeThreshold"))
	}()
	store := realtikvtest.CreateMockStoreAndSetup(t)
	tk1 := testkit.NewTestKit(t, store)
	tk1.Session().GetSessionVars().InitChunkSize = 1
	tk1.Session().GetSessionVars().MaxChunkSize = 1
	tk2 := testkit.NewTestKit(t, store)
	tk1.MustExec("use test")
	tk2.MustExec("use test")
	tk1.MustExec("drop table if exists t")
	tk1.MustExec("create table t(a int primary key, b int)")
	// we need to avoid inserting *literals* into t, so let t2 be the source table.
	tk1.MustExec("create table t2(a int, b int)")
	tk1.MustExec("insert into t2 values (1, 1), (2, 1)")
	require.Nil(t, failpoint.Enable("tikvclient/beforePipelinedFlush", `pause`))
	tk1.MustExec("set session tidb_dml_type = bulk")
	errCh := make(chan error)
	go func() {
		// we expect that this stmt triggers 2 flushes, each containing only 1 row.
		errCh <- tk1.ExecToErr("insert into t select * from t2 order by a")
	}()
	time.Sleep(500 * time.Millisecond)
	tk2.MustExec("insert into t values (1,2)")
	require.Nil(t, failpoint.Disable("tikvclient/beforePipelinedFlush"))
	err := <-errCh
	require.Error(t, err)
	require.True(t, kv.ErrWriteConflict.Equal(err), fmt.Sprintf("error: %s", err))
	require.ErrorContains(t, err, "tableName=test.t")
}