anchor.diff:                  {error, Reason} ->
                       {error, {format, Reason}, State}
               end;
  -        {false, _} ->
  +        {false, _, _} ->
               {error, {format, "Bucket must be allow_mult=true"}, State};
  -        {true, false} ->
  -            {error, {format, "`~p` is not a supported type", [Type]}, State}
  +        {true, false ,_} ->
  +            {error, {format, "Bucket datatype `~p` is not a supported type", [Type]}, State};
  +        {_, _, false} ->
  +            {error, {format, "Operation type is `~p` but  bucket type is `~p`.",
  +                     [OpType, Type]}, State}
       end.
   
   %% @doc process_stream/3 callback. This service does not create any
  
positive.code:         #state{client=C} = State) ->
      {AllowMult, Type} = bucket_type_to_type(B, BType),
      Mod = riak_kv_crdt:to_mod(Type),
      {Key, ReturnKey} = get_key(K),
      case {AllowMult, riak_kv_crdt:supported(Mod)} of
          {true, true} ->
              O = riak_kv_crdt:new(B, Key, Mod),
              Op = riak_pb_dt_codec:decode_operation(Op0, ?MOD_MAP),
              %% erlang_protobuffs encodes as 1/0/undefined
              W = decode_quorum(W0),
              DW = decode_quorum(DW0),
  
negative.code:    undefined;
  get_context(Ctx, true) ->
      Ctx.
  
  mods_match(BucketMod, OpType) ->
      OpMod = riak_kv_crdt:to_mod(OpType),
      OpMod == BucketMod.
  