// PutOperation puts the shard DDL operation into etcd.
PutOperation puts the shard DDL operation into etcd.
// PutOperation puts the shard DDL operation into etcd .
func PutOperation(cli *clientv3.Client, skipDone bool, op Operation, infoModRev int64) (rev int64, putted bool, err error) {
	value, err := op.toJSON()
	if err != nil {
		return 0, false, err
	}
	key := common.ShardDDLOptimismOperationKeyAdapter.Encode(op.Task, op.Source, op.UpSchema, op.UpTable)
	opPut := clientv3.OpPut(key, value)
	cmpsNotExist := make([]clientv3.Cmp, 0, 1)
	cmpsNotDone := make([]clientv3.Cmp, 0, 1)
	cmpsLessRev := make([]clientv3.Cmp, 0, 1)
	if skipDone {
		opDone := op
		opDone.Done = true 
		valueDone, err2 := opDone.toJSON()
		if err2 != nil {
			return 0, false, err2
		}
		cmpsNotExist = append(cmpsNotExist, clientv3util.KeyMissing(key))
		cmpsNotDone = append(cmpsNotDone, clientv3.Compare(clientv3.Value(key), "!=", valueDone))
		cmpsLessRev = append(cmpsLessRev, clientv3.Compare(clientv3.ModRevision(key), "<", infoModRev))
	}
	ctx, cancel := context.WithTimeout(cli.Ctx(), etcdutil.DefaultRequestTimeout)
	defer cancel()
	resp, err := cli.Txn(ctx).If(cmpsNotExist...).Then(opPut).Commit()
	if err != nil {
		return 0, false, err
	} else if resp.Succeeded {
		return resp.Header.Revision, resp.Succeeded, nil
	}
	resp, err = cli.Txn(ctx).If(cmpsNotDone...).Then(opPut).Commit()
	if err != nil {
		return 0, false, err
	} else if resp.Succeeded {
		return resp.Header.Revision, resp.Succeeded, nil
	}
	resp, err = cli.Txn(ctx).If(cmpsLessRev...).Then(opPut).Commit()
	if err != nil {
		return 0, false, err
	}
	return resp.Header.Revision, resp.Succeeded, nil
}
func PutOperation ( cli * clientv3 . Client , skipDone bool , op Operation , infoModRev int64 ) ( rev int64 , putted bool , err error ) { value , err := op . toJSON ( ) if err != nil { return 0 , false , err } key := common . ShardDDLOptimismOperationKeyAdapter . Encode ( op . Task , op . Source , op . UpSchema , op . UpTable ) opPut := clientv3 . OpPut ( key , value ) cmpsNotExist := make ( [ ] clientv3 . Cmp , 0 , 1 ) cmpsNotDone := make ( [ ] clientv3 . Cmp , 0 , 1 ) cmpsLessRev := make ( [ ] clientv3 . Cmp , 0 , 1 ) if skipDone { opDone := op opDone . Done = true valueDone , err2 := opDone . toJSON ( ) if err2 != nil { return 0 , false , err2 } cmpsNotExist = append ( cmpsNotExist , clientv3util . KeyMissing ( key ) ) cmpsNotDone = append ( cmpsNotDone , clientv3 . Compare ( clientv3 . Value ( key ) , " " , valueDone ) ) cmpsLessRev = append ( cmpsLessRev , clientv3 . Compare ( clientv3 . ModRevision ( key ) , " " , infoModRev ) ) } ctx , cancel := context . WithTimeout ( cli . Ctx ( ) , etcdutil . DefaultRequestTimeout ) defer cancel ( ) resp , err := cli . Txn ( ctx ) . If ( cmpsNotExist ... ) . Then ( opPut ) . Commit ( ) if err != nil { return 0 , false , err } else if resp . Succeeded { return resp . Header . Revision , resp . Succeeded , nil } resp , err = cli . Txn ( ctx ) . If ( cmpsNotDone ... ) . Then ( opPut ) . Commit ( ) if err != nil { return 0 , false , err } else if resp . Succeeded { return resp . Header . Revision , resp . Succeeded , nil } resp , err = cli . Txn ( ctx ) . If ( cmpsLessRev ... ) . Then ( opPut ) . Commit ( ) if err != nil { return 0 , false , err } return resp . Header . Revision , resp . Succeeded , nil }
PutOperation puts the shard DDL operation into etcd.
PutOperation puts the shard DDL operation into etcd .
// set `done` to `true`. // txn 1: try to PUT if the key "not exist". // txn 2: try to PUT if the key "the `done`" field is not `true`. // txn 3: try to PUT if the key has less mod revision than info's mod revision, which means this operation is an old one // without this, failed case time series: // 1. dm-master received an old done DDL operation from dm-worker // 2. dm-worker putted a new DDL info into dm-master // 3. dm-master quited before dm-master putted the DDL operation to dm-worker // 4. dm-master restarted and tried to put DDL operation, but found a done one and failed to put // 5. dm-worker didn't receive a DDL operation, will get blocked forever