%% -------------------------------------------------------------------
%%
  Copyright ( c ) 2016 Basho Technologies , Inc.
%%
  This file is provided to you under the Apache License ,
%% Version 2.0 (the "License"); you may not use this file
  except in compliance with the License .   You may obtain
%% a copy of the License at
%%
%%   -2.0
%%
%% Unless required by applicable law or agreed to in writing,
  software distributed under the License is distributed on an
  " AS IS " BASIS , WITHOUT WARRANTIES OR CONDITIONS OF ANY
%% KIND, either express or implied.  See the License for the
%% specific language governing permissions and limitations
%% under the License.
%%
%%-------------------------------------------------------------------

%% @doc Confirm that the batching and queueing behavior in yz_solrq and
%%      related modules works correctly.
-module(yz_solrq_test).
-export([confirm/0]).
-include("yokozuna.hrl").
-include_lib("eunit/include/eunit.hrl").

  TODO : @see cleanup/2 & & cleanup/3 to remove need for creating so many indexes .
-define(INDEX1, <<"solrq_index1">>).
-define(INDEX2, <<"solrq_index2">>).
-define(INDEX3, <<"solrq_index3">>).
-define(INDEX4, <<"solrq_index4">>).
-define(INDEX5, <<"solrq_index5">>).
-define(INDEX6, <<"solrq_index6">>).
-define(INDEX7, <<"solrq_index7">>).
-define(INDEX8, <<"solrq_index8">>).
-define(INDEX9, <<"solrq_index9">>).
-define(INDEX10, <<"solrq_index10">>).
-define(INDEX11, <<"solrq_index11">>).
-define(INDEX12, <<"solrq_index12">>).
-define(BUCKET1, {<<"solrq1">>, <<"solrq_bucket1">>}).
-define(BUCKET2, {<<"solrq2">>, <<"solrq_bucket2">>}).
-define(BUCKET3, {<<"solrq3">>, <<"solrq_bucket3">>}).
-define(BUCKET4, {<<"solrq4">>, <<"solrq_bucket4">>}).
-define(BUCKET5, {<<"solrq5">>, <<"solrq_bucket5">>}).
-define(BUCKET6, {<<"solrq6">>, <<"solrq_bucket6">>}).
-define(BUCKET7, {<<"solrq7">>, <<"solrq_bucket8">>}).
-define(BUCKET8, {<<"solrq8">>, <<"solrq_bucket8">>}).
-define(BUCKET9, {<<"solrq9">>, <<"solrq_bucket9">>}).
-define(BUCKET10, {<<"solrq10">>, <<"solrq_bucket10">>}).
-define(BUCKET11, {<<"solrq11">>, <<"solrq_bucket11">>}).
-define(BUCKET12, {<<"solrq12">>, <<"solrq_bucket12">>}).

-define(BUCKETS, [?BUCKET1, ?BUCKET2, ?BUCKET3, ?BUCKET4, ?BUCKET5,
                  ?BUCKET6, ?BUCKET7, ?BUCKET8, ?BUCKET9, ?BUCKET10,
                  ?BUCKET11, ?BUCKET12]).
-define(INDEXES, [?INDEX1, ?INDEX2, ?INDEX3, ?INDEX4, ?INDEX5,
                  ?INDEX6, ?INDEX7, ?INDEX8, ?INDEX9, ?INDEX10,
                  ?INDEX11, ?INDEX12]).

-define(SOLRQ_DELAYMS_MAX, 3000).
-define(SOLRQ_BATCH_MIN_SETTING, 4).
-define(SOLRQ_BATCH_MAX_SETTING, 8).
-define(MELT_RESET_REFRESH, 1000).
-define(SOLRQ_HWM_SETTING, 20).
-define(RING_SIZE, 8).
-define(CONFIG,
        [{yokozuna,
          [{enabled, true},
           {?SOLRQ_BATCH_FLUSH_INTERVAL, ?SOLRQ_DELAYMS_MAX},
           {?SOLRQ_BATCH_MIN, ?SOLRQ_BATCH_MIN_SETTING},
           {?SOLRQ_BATCH_MAX, ?SOLRQ_BATCH_MAX_SETTING},
           {?SOLRQ_HWM, 1000},
           {?ERR_THRESH_FAIL_COUNT, 1},
           {?ERR_THRESH_RESET_INTERVAL, ?MELT_RESET_REFRESH},
           {?SOLRQ_DRAIN_ENABLE, true},
           {anti_entropy, {off, []}}
          ]},
            {riak_core, [{ring_creation_size, ?RING_SIZE}]}]).

-compile(export_all).

confirm() ->
    Cluster = yz_rt:prepare_cluster(1, ?CONFIG),
    [PBConn|_] = PBConns = yz_rt:open_pb_conns(Cluster),

    ok = yz_rt:create_indexed_bucket(PBConn, Cluster, ?BUCKET1, ?INDEX1),
    confirm_batch_size(Cluster, PBConn, ?BUCKET1, ?INDEX1),
    [confirm_hwm(Cluster, PBConn, ?BUCKET1, ?INDEX1, HWM) || HWM <- lists:seq(0, 10)],

    ok = yz_rt:create_indexed_bucket(PBConn, Cluster, ?BUCKET2, ?INDEX2),
    confirm_draining(Cluster, PBConn, ?BUCKET2, ?INDEX2),

    confirm_drain_fsm_failure(Cluster),
    confirm_drain_fsm_timeout(Cluster),
    confirm_drain_fsm_kill(Cluster),

    %% Buckets and Indexes 3 - 12.
    [ok = yz_rt:create_indexed_bucket(PBConn, Cluster, Bucket, Index)
     || {Bucket, Index} <- lists:zip(lists:nthtail(2, ?BUCKETS),
                                   lists:nthtail(2, ?INDEXES))],

    %% confirm_requeue_undelivered must be last since it installs an interrupt
    %% that intentionally causes failures
    confirm_requeue_undelivered(Cluster, PBConn, ?BUCKET3, ?INDEX3),

    confirm_no_contenttype_data(Cluster, PBConn, ?BUCKET4, ?INDEX4),
    confirm_purge_strategy(Cluster, PBConn),

    yz_rt:close_pb_conns(PBConns),
    pass.

confirm_drain_fsm_failure(Cluster) ->
    lager:info("Starting confirm_drain_fsm_failure"),
    yz_stat:reset(),
    try
        yz_rt:load_intercept_code(Cluster),
        yz_rt:add_intercepts(Cluster, yz_solrq_drain_fsm, [{{prepare, 2}, prepare_crash}]),
  drain solrqs and wait until the drain failure stats are touched
        yz_rt:drain_solrqs(Cluster),
        yz_rt:wait_until(Cluster, fun check_drain_failure_stats/1),

        lager:info("confirm_drain_fsm_failure ok")
    after
        yz_rt:add_intercepts(Cluster, yz_solrq_drain_fsm, [{{prepare, 2}, prepare_orig}])
    end.

check_drain_failure_stats(Node) ->
    Stats = rpc:call(Node, yz_stat, get_stats, []),

    QDrainFail = proplists:get_value(yz_stat:stat_name([queue, drain, fail]), Stats),
    QDrainFailCount = proplists:get_value(count, QDrainFail),
    QDrainFailOne = proplists:get_value(one, QDrainFail),

    Pairs = [
        {queue_drain_fail_count, QDrainFailCount, '>', 0},
        {queue_drain_fail_one, QDrainFailOne, '>', 0}
    ],
    yz_rt:check_stat_values(Stats, Pairs).

confirm_drain_fsm_timeout(Cluster) ->
    lager:info("Starting confirm_drain_fsm_timeout"),
    yz_stat:reset(),
    [rpc:call(
        Node, application, set_env, [?YZ_APP_NAME, ?SOLRQ_DRAIN_TIMEOUT, 250])
            || Node <- Cluster],
    try
        yz_rt:load_intercept_code(Cluster),
        yz_rt:add_intercepts(Cluster, yz_solrq_drain_fsm, [{{resume_workers, 1}, resume_workers_sleep_1s}]),
        yz_rt:drain_solrqs(Cluster),
        yz_rt:wait_until(Cluster, fun check_drain_timeout_stats/1),

        lager:info("confirm_drain_fsm_timeout ok")
    after
        yz_rt:add_intercepts(Cluster, yz_solrq_drain_fsm, [{{resume_workers, 1}, resume_workers_orig}]),
        [rpc:call(
            Node, application, set_env, [?YZ_APP_NAME, ?SOLRQ_DRAIN_TIMEOUT, 60000])
                || Node <- Cluster]
    end.

check_drain_timeout_stats(Node) ->
    Stats = rpc:call(Node, yz_stat, get_stats, []),

    QDrainTimeout = proplists:get_value(yz_stat:stat_name([queue, drain, timeout]), Stats),
    QDrainTimeoutCount = proplists:get_value(count, QDrainTimeout),
    QDrainTimeoutOne = proplists:get_value(one, QDrainTimeout),

    Pairs = [
        {queue_drain_timeout_count, QDrainTimeoutCount, '>', 0},
        {queue_drain_timeout_one, QDrainTimeoutOne, '>', 0}
    ],
    yz_rt:check_stat_values(Stats, Pairs).

confirm_drain_fsm_kill(Cluster) ->
    lager:info("Starting confirm_drain_fsm_kill"),
    [rpc:call(
        Node, application, set_env, [?YZ_APP_NAME, ?SOLRQ_DRAIN_TIMEOUT, 10])
        || Node <- Cluster],
    %% technically not needed for this test (because the cancel intercept will
    %% just return timeout), but added for completeness
    [rpc:call(
        Node, application, set_env, [?YZ_APP_NAME, ?SOLRQ_DRAIN_CANCEL_TIMEOUT, 10])
        || Node <- Cluster],
    try
        yz_test_listener:start(),
        yz_rt:load_intercept_code(Cluster),
        yz_rt:add_intercepts(Cluster, yz_solrq_drain_fsm, [{{resume_workers, 1}, resume_workers_sleep_1s},
                                                           {{cancel, 2}, cancel_timeout}]),
        yz_rt:add_intercepts(Cluster, yz_solrq_drain_mgr, [{{unlink_and_kill, 2}, count_unlink_and_kill}]),
        yz_rt:drain_solrqs(Cluster),
        yz_rt:wait_until(Cluster, fun check_drain_cancel_timeout_stats/1),

        ?assertEqual(1, length(yz_test_listener:messages())),

        lager:info("confirm_drain_fsm_kill ok")
    after
        yz_rt:add_intercepts(Cluster, yz_solrq_drain_fsm, [{{resume_workers, 1}, resume_workers_orig},
                                                           {{cancel, 2}, cancel_orig}]),
        yz_rt:add_intercepts(Cluster, yz_solrq_drain_mgr, [{{unlink_and_kill, 2}, unlink_and_kill_orig}]),
        yz_test_listener:stop(),
        [rpc:call(
            Node, application, set_env, [?YZ_APP_NAME, ?SOLRQ_DRAIN_TIMEOUT, 60000])
            || Node <- Cluster],
        [rpc:call(
            Node, application, set_env, [?YZ_APP_NAME, ?SOLRQ_DRAIN_CANCEL_TIMEOUT, 5000])
            || Node <- Cluster]
    end.

check_drain_cancel_timeout_stats(Node) ->
    Stats = rpc:call(Node, yz_stat, get_stats, []),

    QDrainTimeout = proplists:get_value(yz_stat:stat_name([queue, drain, cancel, timeout]), Stats),
    QDrainTimeoutCount = proplists:get_value(count, QDrainTimeout),
    QDrainTimeoutOne = proplists:get_value(one, QDrainTimeout),

    Pairs = [
        {queue_drain_cancel_timeout_count, QDrainTimeoutCount, '>', 0},
        {queue_drain_cancel_timeout_one, QDrainTimeoutOne, '>', 0}
    ],
    yz_rt:check_stat_values(Stats, Pairs).


confirm_batch_size(Cluster, PBConn, BKey, Index) ->
    lager:info("Starting confirm_batch_size"),
  First , put one less than the min batch size and expect that there are no
    %% search results (because the index operations are queued).
    Count = ?SOLRQ_BATCH_MIN_SETTING - 1,
    Count = put_objects(PBConn, BKey, Count),
    yz_rt:commit(Cluster, Index),
    verify_search_count(PBConn, Index, 0),

  Now , put one more and expect that all of them have been indexed ( because
  the solrq_batch_min has been reached ) and therefore show up in search
    %% results.
    1 = put_objects(PBConn, BKey, 1),
    yz_rt:commit(Cluster, Index),
    verify_search_count(PBConn, Index, ?SOLRQ_BATCH_MIN_SETTING),

  Finally , put another batch of one less than solrq_batch_min , but this
    %% time wait until the data has been flushed, and then verify that we
  have waited at least SOLRQ_DELAYMS_MAX ms .
    Count = put_objects(PBConn, BKey, Count),
    yz_rt:commit(Cluster, Index),
    verify_search_count(PBConn, Index, ?SOLRQ_BATCH_MIN_SETTING),
    StartMS = current_ms(),
    wait_until_search_count(PBConn, Index, ?SOLRQ_BATCH_MIN_SETTING + Count),
    Condition = gteq(current_ms() - StartMS, ?SOLRQ_DELAYMS_MAX),
    case Condition of
        false -> lager:error("Flush interval is less than ~pms", [?SOLRQ_DELAYMS_MAX]);
        true -> ok
    end,
    ?assertEqual(Condition, true),
    lager:info("confirm_batch_size ok"),
    ok.

confirm_hwm(Cluster, PBConn, Bucket, Index, HWM) ->
    lager:info("Starting confirm_hwm"),
    yz_rt:drain_solrqs(Cluster),
    {OldMin, OldMax, OldDelay} = set_index(Cluster, Index, 1, 100, 100),
    try
        yz_rt:load_intercept_code(Cluster),
        yz_rt:intercept_index_batch(Cluster, index_batch_returns_other_error),
        yz_rt:set_hwm(Cluster, HWM),
        yz_rt:set_purge_strategy(Cluster, ?PURGE_NONE),
        ?assertEqual(HWM, put_objects(PBConn, Bucket, HWM + 1))
    after
        yz_rt:intercept_index_batch(Cluster, index_batch_call_orig),
        yz_rt:set_purge_strategy(Cluster, ?PURGE_ONE),
        yz_rt:set_hwm(Cluster, 10000),
        yz_rt:set_index(Cluster, Index, OldMin, OldMax, OldDelay)
    end,
    lager:info("confirm_hwm ok for HWM ~p", [HWM]),
    ok.

-spec gteq(number(), number()) -> boolean().
gteq(A, B) -> A >= B.

confirm_draining(Cluster, PBConn, Bucket, Index) ->
    lager:info("Starting confirm_draining"),
    Count = ?SOLRQ_BATCH_MIN_SETTING - 1,
    Count = put_objects(PBConn, Bucket, Count),
    yz_rt:commit(Cluster, Index),
    verify_search_count(PBConn, Index, 0),
    yz_rt:drain_solrqs(Cluster),
    yz_rt:commit(Cluster, Index),
    verify_search_count(PBConn, Index, Count),
    lager:info("confirm_draining ok"),
    ok.

confirm_requeue_undelivered([Node|_] = Cluster, PBConn, BKey, Index) ->
    lager:info("Starting confirm_requeue_undelivered"),
    yz_rt:load_intercept_code(Node),
    yz_rt:intercept_index_batch(Node, index_batch_returns_other_error),

    Count = ?SOLRQ_BATCH_MIN_SETTING,
    Count = put_objects(PBConn, BKey, Count),
    yz_rt:commit(Cluster, Index),

    %% Because the index_batch_returns_other_error intercept simulates a Solr
    %% failure, none of the objects should have been indexed at this point.
    verify_search_count(PBConn, Index, 0),

    %% Now, if we replace the intercept with one that just calls the original
    %% function, the undelivered objects will be requeued and should succeed
    %% (assuming that the fuse has been blown and reset).
    yz_rt:intercept_index_batch(Node, index_batch_call_orig),
    timer:sleep(?MELT_RESET_REFRESH + 1000), %% wait for fuse reset
    yz_rt:drain_solrqs(Node),
    verify_search_count(PBConn, Index, Count),
    lager:info("confirm_requeue_undelivered ok"),
    ok.

confirm_no_contenttype_data(Cluster, PBConn, BKey, Index) ->
    lager:info("Starting confirm_no_contenttype_data"),
    yz_rt:set_index(Cluster, Index, 1, 100, 100),
    Count = 1,
    Count = put_no_contenttype_objects(PBConn, BKey, Count),
    yz_rt:commit(Cluster, Index),
    verify_search_count(PBConn, Index, 1),
    lager:info("confirm_no_contenttype_data ok"),
    ok.

confirm_purge_strategy(Cluster, PBConn) ->
    lager:info("Starting confirm_purge_strategy"),
    confirm_purge_one_strategy(Cluster, PBConn,
                               {?BUCKET5, ?INDEX5}),
    confirm_purge_idx_strategy(Cluster, PBConn,
                               {?BUCKET7, ?INDEX7}),
    confirm_purge_none_strategy(Cluster, PBConn,
                                {?BUCKET11, ?INDEX11}),
    ok.

confirm_purge_one_strategy(Cluster, PBConn, Bucket1Index1) ->
    PurgeResults = do_purge(Cluster, PBConn, Bucket1Index1, ?PURGE_ONE),
    check_one_purged(PurgeResults),
    lager:info("confirm_purge_one_strategy ok"),
    ok.

check_one_purged({Index1Written, Index1SearchResults} = TestResults) ->
    Condition = first_purged(Index1Written, Index1SearchResults),
    case Condition of
        false ->
            lager:error("check_one_purged error: ~p", [TestResults]);
        _ -> ok
    end,
    ?assertEqual(Condition, true),
    ok.

confirm_purge_idx_strategy(Cluster, PBConn, Bucket1Index1) ->
    PurgeResults = do_purge(Cluster, PBConn, Bucket1Index1, ?PURGE_IDX),
    check_idx_purged(PurgeResults),
    lager:info("confirm_purge_idx_strategy ok"),
    ok.

check_idx_purged({[_K1, _K2, _K3, _K4, K5] = _Index1Written, Index1SearchResults} = TestResults) ->
    %%
  Note the second condition , because we wrote to Index1
    %% but that was the purge trigger, so the last entry will
    %% NOT have been purged, of that indexq was the one chosen.
  Otherwise , it was the second indexq , and nothing should
    %% have been pending for that indexq, so they all get deleted.
    %%
    Condition = equal(Index1SearchResults, [K5]),
    case Condition of
        false ->
            lager:error("check_idx_purged error: ~p", [TestResults]);
        _ -> ok
    end,
    ?assertEqual(Condition, true),
    ok.

confirm_purge_none_strategy(Cluster, PBConn, Bucket1Index1) ->
    PurgeResults = do_purge(Cluster, PBConn, Bucket1Index1,
                            ?PURGE_NONE),
    check_none_purged(PurgeResults),
    lager:info("confirm_purge_none_strategy ok"),
    ok.

check_none_purged({Index1Written, Index1SearchResults} = TestResults) ->
    Condition =
        equal(Index1Written, Index1SearchResults),
    case Condition of
        false ->
            lager:error("check_none_purged error: ~p", [TestResults]);
        _ -> ok
    end,
    ?assertEqual(Condition, true),
    ok.

equal(Written, Searched) ->
    lists:sort(Written) == lists:sort(Searched).

first_purged([_H|T] = _Written, Searched) ->
    lists:sort(T) == lists:sort(Searched).

%%
  All the purge tests use this function to load up 2 indexqs
%% in the first solrq (yz_solrq_001 -- we choose it arbitrarily,
%% but we only need one to test the different purge scenarios)
  up to the HWM .   We do this by setting an intercept so that
  the call to Solr fails for all of these calls , which will trigger
%% the fuses to melt for both of these indices.
%%
  Once the fuses for the indexqs are blown , we send one message
  using the first index .   This will trigger a purge , whose behavior
%% is determined by the supplied purge strategy.
%%
  We then restore the original Solr behavior by reverting the intercept ,
%% and we drain the queues to push through any data that had been enqueued
%% but not pushed to Solr.  We return the list of keys we wrote to
%% Riak and the search results for each index, which is compared outside
%% of this function.
%%
do_purge([Node|_] = Cluster, PBConn,
         {Bucket, Index},
         PurgeStrategy) ->
    yz_rt:set_purge_strategy(Cluster, PurgeStrategy),
    yz_rt:set_index(Cluster, Index, 1, 100, 99999),
    yz_rt:set_hwm(Cluster, 4),
    TargetPartition = 1096126227998177188652763624537212264741949407232,
    %%
    %% Find a list of representative keys.
    %% Each representative in the list is a unique key
    %% that hashes to the target partition.
    %%
    RepresentativeKeys = find_representative_bkeys(TargetPartition, Index, Bucket),
    Key1 = lists:nth(1, RepresentativeKeys),
    Key2 = lists:nth(2, RepresentativeKeys),
    Key3 = lists:nth(3, RepresentativeKeys),
    Key4 = lists:nth(4, RepresentativeKeys),
    Key5 = lists:nth(5, RepresentativeKeys),
    try
        yz_rt:load_intercept_code(Node),
        yz_rt:intercept_index_batch(Node, index_batch_returns_other_error),
        %%
        %% Send messages through the index on the solrq, which
        %% will trip the fuse; however
        %% because fuse blown events are handled asynchronously,
        %% we need to wait until the solrq is blown.
        %%
        [Key1] = put_bkey_objects(PBConn, [Key1]),
        [Key2] = put_bkey_objects(PBConn, [Key2]),
        [Key3] = put_bkey_objects(PBConn, [Key3]),
        [Key4] = put_bkey_objects(PBConn, [Key4]),
        yz_rt:wait_until_fuses_blown(Node, TargetPartition, [Index]),
        %%
        %% At this point, the target solrq corresponding
  to { TargetPartition , Index1 } , should be blown .
  Send one more message through the Indexq , which
        %% will trigger a purge.
        %%
        yz_rt:set_index(Cluster, Index, 1, 100, 99999),
        F = fun() ->
            [Key5] = put_bkey_objects(PBConn, [Key5])
        end,
        case PurgeStrategy of
            %% If testing PURGE_NONE, all vnodes are blocked
            %% spawn the call so we don't block the test.of
            %% Note that this will result in an error showing up
            %% in the test log, but that's ok as we expect it.
            ?PURGE_NONE ->
                spawn(F);
            _ ->
                F()
        end
    after
        %%
        %% Revert the intercept, and drain, giving time for the
        %% fuse to reset.  Commit to Solr so that we can run a query.
        %%
        yz_rt:intercept_index_batch(Node, index_batch_call_orig),
        yz_rt:wait_until_fuses_reset(Node, TargetPartition, [Index]),
        yz_rt:drain_solrqs(Node),
        yz_rt:commit(Cluster, Index)
    end,
    %%
    %% Return the search results for the index.
  The first list is the set of bkeys we wrote .
  The second list is the set that are available for search .
    %%
    KeysFound = search_bkeys(PBConn, Index),
    {[Key1, Key2, Key3, Key4, Key5], KeysFound}.

-spec search_bkeys(pid(), index_name()) -> [bkey()].
search_bkeys(PBConn, Index) ->
    {ok, {search_results, SearchResults, _Score, _NumFound}} =
        riakc_pb_socket:search(PBConn, Index, <<"*:*">>),
    lists:map(
        fun({_Index, Fields}) ->
                {{proplists:get_value(<<"_yz_rt">>, Fields),
                  proplists:get_value(<<"_yz_rb">>, Fields)},
                 proplists:get_value(<<"_yz_rk">>, Fields)}
        end,
        SearchResults).

-spec find_representative_bkeys(p(),
                                index_name(),
                                bucket()) -> [bkey()].
find_representative_bkeys(Partition, Index, Bucket) ->
    Solrq = yz_solrq:worker_regname(Index, Partition),
    Representatives = find_representatives(Index, Bucket),
    {ok, BKeys} = dict:find(Solrq, Representatives),
    BKeys.

-spec find_representatives(index_name(), bucket()) -> dict().
find_representatives(Index, Bucket) ->
    BKeys =
        lists:map(
          fun(I) ->
                  {Bucket, erlang:list_to_binary(erlang:integer_to_list(I))}
          end,
          lists:seq(1, 1000)),
    yz_rt:find_representatives(Index, BKeys, ?RING_SIZE).

-spec put_no_contenttype_objects(pid(), bucket(), non_neg_integer()) -> non_neg_integer().
put_no_contenttype_objects(PBConn, Bucket, Count) ->
    put_objects(PBConn, Bucket, Count, undefined).

-spec put_objects(pid(), bucket(), non_neg_integer()) -> non_neg_integer().
put_objects(PBConn, Bucket, Count) ->
    put_objects(PBConn, Bucket, Count, "text/plain").

-spec put_objects(pid(), bucket(), non_neg_integer(), string()|undefined) -> non_neg_integer().
put_objects(PBConn, Bucket, Count, ContentType) ->
    %% Using the same key for every object ensures that they all hash to the
    %% same yz_solrq_worker and the batching is therefore predictable.
    Key = <<"same_key_for_everyone">>,
    RandVals = [yz_rt:random_binary(16) || _ <- lists:seq(1, Count)],
    Objects = [case ContentType of
                   undefined ->
                       riakc_obj:new(Bucket, Key, Val);
                   _ ->
                       riakc_obj:new(Bucket, Key, Val, ContentType)
               end || Val <- RandVals],
    lager:info("Writing ~p objects to Riak...", [length(Objects)]),
    Results = [riakc_pb_socket:put(PBConn, Obj, [return_head, {timeout, 1000}]) || Obj <- Objects],
    NoWritten = length(lists:filter(fun({Result, _}) ->
                                        ok =:= Result
                                    end,
                                    Results)),
    lager:info("Wrote ~p objects to Riak...", [NoWritten]),
    NoWritten.

-spec put_bkey_objects(pid(), [bkey()]) -> [bkey()].
put_bkey_objects(PBConn, BKeys) ->
    BKeyObjects = [{{Bucket, Key},
                    riakc_obj:new(Bucket, Key, Key, "text/plain")}
                   || {Bucket, Key} <- BKeys],
    lager:info("Writing ~p bkeys ~p to Riak...", [length(BKeys), BKeys]),
    Results = [{BKey, riakc_pb_socket:put(PBConn, Obj,
                                          [return_head, {timeout, 1000}])}
               || {BKey, Obj} <- BKeyObjects],
    %%lager:info("Results: ~p", [Results]),
    lists:map(
      fun({BKey, {_Result, _}}) -> BKey end,
      lists:filter(
          fun({BKey, {Result, _}}) ->
              case Result of
                  ok -> true;
                  _ ->
                      lager:info("The result of a put was not ok.  BKey: ~p Result: ~p", [BKey, Result]),
                      false
              end
          end,
          Results)).

verify_search_count(PBConn, Index, Count) ->
    {ok, {search_results, _R, _Score, Found}} =
        riakc_pb_socket:search(PBConn, Index, <<"*:*">>),
    ?assertEqual(Count, Found).

wait_until_search_count(PBConn, Index, Count) ->
    F = fun() ->
        {ok, {search_results, _R, _Score, Found}} =
            riakc_pb_socket:search(PBConn, Index, <<"*:*">>),
        Found =:= Count
        end,
    rt:wait_until(F).

current_ms() ->
    {MegaSecs, Secs, MicroSecs} = os:timestamp(),
    (MegaSecs * 1000000 + Secs) * 1000 + round(MicroSecs / 1000).

set_index(Cluster, Index, Min, Max, Delay) ->
    {[[{_WorkerId, {ok, {OldMin, OldMax, OldDelay}}}|_]], _} =
        yz_rt:set_index(Cluster, Index, Min, Max, Delay),
    {OldMin, OldMax, OldDelay}.


 -------------------------------------------------------------------


 Version 2.0 (the "License"); you may not use this file
 a copy of the License at

   -2.0

 Unless required by applicable law or agreed to in writing,
 KIND, either express or implied.  See the License for the
 specific language governing permissions and limitations
 under the License.

-------------------------------------------------------------------
 @doc Confirm that the batching and queueing behavior in yz_solrq and
      related modules works correctly.
 Buckets and Indexes 3 - 12.
 confirm_requeue_undelivered must be last since it installs an interrupt
 that intentionally causes failures
 technically not needed for this test (because the cancel intercept will
 just return timeout), but added for completeness
 search results (because the index operations are queued).
 results.
 time wait until the data has been flushed, and then verify that we
 Because the index_batch_returns_other_error intercept simulates a Solr
 failure, none of the objects should have been indexed at this point.
 Now, if we replace the intercept with one that just calls the original
 function, the undelivered objects will be requeued and should succeed
 (assuming that the fuse has been blown and reset).
 wait for fuse reset

 but that was the purge trigger, so the last entry will
 NOT have been purged, of that indexq was the one chosen.
 have been pending for that indexq, so they all get deleted.


 in the first solrq (yz_solrq_001 -- we choose it arbitrarily,
 but we only need one to test the different purge scenarios)
 the fuses to melt for both of these indices.

 is determined by the supplied purge strategy.

 and we drain the queues to push through any data that had been enqueued
 but not pushed to Solr.  We return the list of keys we wrote to
 Riak and the search results for each index, which is compared outside
 of this function.


 Find a list of representative keys.
 Each representative in the list is a unique key
 that hashes to the target partition.


 Send messages through the index on the solrq, which
 will trip the fuse; however
 because fuse blown events are handled asynchronously,
 we need to wait until the solrq is blown.


 At this point, the target solrq corresponding
 will trigger a purge.

 If testing PURGE_NONE, all vnodes are blocked
 spawn the call so we don't block the test.of
 Note that this will result in an error showing up
 in the test log, but that's ok as we expect it.

 Revert the intercept, and drain, giving time for the
 fuse to reset.  Commit to Solr so that we can run a query.


 Return the search results for the index.

 Using the same key for every object ensures that they all hash to the
 same yz_solrq_worker and the batching is therefore predictable.
lager:info("Results: ~p", [Results]),
  Copyright ( c ) 2016 Basho Technologies , Inc.
  This file is provided to you under the Apache License ,
  except in compliance with the License .   You may obtain
  software distributed under the License is distributed on an
  " AS IS " BASIS , WITHOUT WARRANTIES OR CONDITIONS OF ANY

-module(yz_solrq_test).
-export([confirm/0]).
-include("yokozuna.hrl").
-include_lib("eunit/include/eunit.hrl").

  TODO : @see cleanup/2 & & cleanup/3 to remove need for creating so many indexes .
-define(INDEX1, <<"solrq_index1">>).
-define(INDEX2, <<"solrq_index2">>).
-define(INDEX3, <<"solrq_index3">>).
-define(INDEX4, <<"solrq_index4">>).
-define(INDEX5, <<"solrq_index5">>).
-define(INDEX6, <<"solrq_index6">>).
-define(INDEX7, <<"solrq_index7">>).
-define(INDEX8, <<"solrq_index8">>).
-define(INDEX9, <<"solrq_index9">>).
-define(INDEX10, <<"solrq_index10">>).
-define(INDEX11, <<"solrq_index11">>).
-define(INDEX12, <<"solrq_index12">>).
-define(BUCKET1, {<<"solrq1">>, <<"solrq_bucket1">>}).
-define(BUCKET2, {<<"solrq2">>, <<"solrq_bucket2">>}).
-define(BUCKET3, {<<"solrq3">>, <<"solrq_bucket3">>}).
-define(BUCKET4, {<<"solrq4">>, <<"solrq_bucket4">>}).
-define(BUCKET5, {<<"solrq5">>, <<"solrq_bucket5">>}).
-define(BUCKET6, {<<"solrq6">>, <<"solrq_bucket6">>}).
-define(BUCKET7, {<<"solrq7">>, <<"solrq_bucket8">>}).
-define(BUCKET8, {<<"solrq8">>, <<"solrq_bucket8">>}).
-define(BUCKET9, {<<"solrq9">>, <<"solrq_bucket9">>}).
-define(BUCKET10, {<<"solrq10">>, <<"solrq_bucket10">>}).
-define(BUCKET11, {<<"solrq11">>, <<"solrq_bucket11">>}).
-define(BUCKET12, {<<"solrq12">>, <<"solrq_bucket12">>}).

-define(BUCKETS, [?BUCKET1, ?BUCKET2, ?BUCKET3, ?BUCKET4, ?BUCKET5,
                  ?BUCKET6, ?BUCKET7, ?BUCKET8, ?BUCKET9, ?BUCKET10,
                  ?BUCKET11, ?BUCKET12]).
-define(INDEXES, [?INDEX1, ?INDEX2, ?INDEX3, ?INDEX4, ?INDEX5,
                  ?INDEX6, ?INDEX7, ?INDEX8, ?INDEX9, ?INDEX10,
                  ?INDEX11, ?INDEX12]).

-define(SOLRQ_DELAYMS_MAX, 3000).
-define(SOLRQ_BATCH_MIN_SETTING, 4).
-define(SOLRQ_BATCH_MAX_SETTING, 8).
-define(MELT_RESET_REFRESH, 1000).
-define(SOLRQ_HWM_SETTING, 20).
-define(RING_SIZE, 8).
-define(CONFIG,
        [{yokozuna,
          [{enabled, true},
           {?SOLRQ_BATCH_FLUSH_INTERVAL, ?SOLRQ_DELAYMS_MAX},
           {?SOLRQ_BATCH_MIN, ?SOLRQ_BATCH_MIN_SETTING},
           {?SOLRQ_BATCH_MAX, ?SOLRQ_BATCH_MAX_SETTING},
           {?SOLRQ_HWM, 1000},
           {?ERR_THRESH_FAIL_COUNT, 1},
           {?ERR_THRESH_RESET_INTERVAL, ?MELT_RESET_REFRESH},
           {?SOLRQ_DRAIN_ENABLE, true},
           {anti_entropy, {off, []}}
          ]},
            {riak_core, [{ring_creation_size, ?RING_SIZE}]}]).

-compile(export_all).

confirm() ->
    Cluster = yz_rt:prepare_cluster(1, ?CONFIG),
    [PBConn|_] = PBConns = yz_rt:open_pb_conns(Cluster),

    ok = yz_rt:create_indexed_bucket(PBConn, Cluster, ?BUCKET1, ?INDEX1),
    confirm_batch_size(Cluster, PBConn, ?BUCKET1, ?INDEX1),
    [confirm_hwm(Cluster, PBConn, ?BUCKET1, ?INDEX1, HWM) || HWM <- lists:seq(0, 10)],

    ok = yz_rt:create_indexed_bucket(PBConn, Cluster, ?BUCKET2, ?INDEX2),
    confirm_draining(Cluster, PBConn, ?BUCKET2, ?INDEX2),

    confirm_drain_fsm_failure(Cluster),
    confirm_drain_fsm_timeout(Cluster),
    confirm_drain_fsm_kill(Cluster),

    [ok = yz_rt:create_indexed_bucket(PBConn, Cluster, Bucket, Index)
     || {Bucket, Index} <- lists:zip(lists:nthtail(2, ?BUCKETS),
                                   lists:nthtail(2, ?INDEXES))],

    confirm_requeue_undelivered(Cluster, PBConn, ?BUCKET3, ?INDEX3),

    confirm_no_contenttype_data(Cluster, PBConn, ?BUCKET4, ?INDEX4),
    confirm_purge_strategy(Cluster, PBConn),

    yz_rt:close_pb_conns(PBConns),
    pass.

confirm_drain_fsm_failure(Cluster) ->
    lager:info("Starting confirm_drain_fsm_failure"),
    yz_stat:reset(),
    try
        yz_rt:load_intercept_code(Cluster),
        yz_rt:add_intercepts(Cluster, yz_solrq_drain_fsm, [{{prepare, 2}, prepare_crash}]),
  drain solrqs and wait until the drain failure stats are touched
        yz_rt:drain_solrqs(Cluster),
        yz_rt:wait_until(Cluster, fun check_drain_failure_stats/1),

        lager:info("confirm_drain_fsm_failure ok")
    after
        yz_rt:add_intercepts(Cluster, yz_solrq_drain_fsm, [{{prepare, 2}, prepare_orig}])
    end.

check_drain_failure_stats(Node) ->
    Stats = rpc:call(Node, yz_stat, get_stats, []),

    QDrainFail = proplists:get_value(yz_stat:stat_name([queue, drain, fail]), Stats),
    QDrainFailCount = proplists:get_value(count, QDrainFail),
    QDrainFailOne = proplists:get_value(one, QDrainFail),

    Pairs = [
        {queue_drain_fail_count, QDrainFailCount, '>', 0},
        {queue_drain_fail_one, QDrainFailOne, '>', 0}
    ],
    yz_rt:check_stat_values(Stats, Pairs).

confirm_drain_fsm_timeout(Cluster) ->
    lager:info("Starting confirm_drain_fsm_timeout"),
    yz_stat:reset(),
    [rpc:call(
        Node, application, set_env, [?YZ_APP_NAME, ?SOLRQ_DRAIN_TIMEOUT, 250])
            || Node <- Cluster],
    try
        yz_rt:load_intercept_code(Cluster),
        yz_rt:add_intercepts(Cluster, yz_solrq_drain_fsm, [{{resume_workers, 1}, resume_workers_sleep_1s}]),
        yz_rt:drain_solrqs(Cluster),
        yz_rt:wait_until(Cluster, fun check_drain_timeout_stats/1),

        lager:info("confirm_drain_fsm_timeout ok")
    after
        yz_rt:add_intercepts(Cluster, yz_solrq_drain_fsm, [{{resume_workers, 1}, resume_workers_orig}]),
        [rpc:call(
            Node, application, set_env, [?YZ_APP_NAME, ?SOLRQ_DRAIN_TIMEOUT, 60000])
                || Node <- Cluster]
    end.

check_drain_timeout_stats(Node) ->
    Stats = rpc:call(Node, yz_stat, get_stats, []),

    QDrainTimeout = proplists:get_value(yz_stat:stat_name([queue, drain, timeout]), Stats),
    QDrainTimeoutCount = proplists:get_value(count, QDrainTimeout),
    QDrainTimeoutOne = proplists:get_value(one, QDrainTimeout),

    Pairs = [
        {queue_drain_timeout_count, QDrainTimeoutCount, '>', 0},
        {queue_drain_timeout_one, QDrainTimeoutOne, '>', 0}
    ],
    yz_rt:check_stat_values(Stats, Pairs).

confirm_drain_fsm_kill(Cluster) ->
    lager:info("Starting confirm_drain_fsm_kill"),
    [rpc:call(
        Node, application, set_env, [?YZ_APP_NAME, ?SOLRQ_DRAIN_TIMEOUT, 10])
        || Node <- Cluster],
    [rpc:call(
        Node, application, set_env, [?YZ_APP_NAME, ?SOLRQ_DRAIN_CANCEL_TIMEOUT, 10])
        || Node <- Cluster],
    try
        yz_test_listener:start(),
        yz_rt:load_intercept_code(Cluster),
        yz_rt:add_intercepts(Cluster, yz_solrq_drain_fsm, [{{resume_workers, 1}, resume_workers_sleep_1s},
                                                           {{cancel, 2}, cancel_timeout}]),
        yz_rt:add_intercepts(Cluster, yz_solrq_drain_mgr, [{{unlink_and_kill, 2}, count_unlink_and_kill}]),
        yz_rt:drain_solrqs(Cluster),
        yz_rt:wait_until(Cluster, fun check_drain_cancel_timeout_stats/1),

        ?assertEqual(1, length(yz_test_listener:messages())),

        lager:info("confirm_drain_fsm_kill ok")
    after
        yz_rt:add_intercepts(Cluster, yz_solrq_drain_fsm, [{{resume_workers, 1}, resume_workers_orig},
                                                           {{cancel, 2}, cancel_orig}]),
        yz_rt:add_intercepts(Cluster, yz_solrq_drain_mgr, [{{unlink_and_kill, 2}, unlink_and_kill_orig}]),
        yz_test_listener:stop(),
        [rpc:call(
            Node, application, set_env, [?YZ_APP_NAME, ?SOLRQ_DRAIN_TIMEOUT, 60000])
            || Node <- Cluster],
        [rpc:call(
            Node, application, set_env, [?YZ_APP_NAME, ?SOLRQ_DRAIN_CANCEL_TIMEOUT, 5000])
            || Node <- Cluster]
    end.

check_drain_cancel_timeout_stats(Node) ->
    Stats = rpc:call(Node, yz_stat, get_stats, []),

    QDrainTimeout = proplists:get_value(yz_stat:stat_name([queue, drain, cancel, timeout]), Stats),
    QDrainTimeoutCount = proplists:get_value(count, QDrainTimeout),
    QDrainTimeoutOne = proplists:get_value(one, QDrainTimeout),

    Pairs = [
        {queue_drain_cancel_timeout_count, QDrainTimeoutCount, '>', 0},
        {queue_drain_cancel_timeout_one, QDrainTimeoutOne, '>', 0}
    ],
    yz_rt:check_stat_values(Stats, Pairs).


confirm_batch_size(Cluster, PBConn, BKey, Index) ->
    lager:info("Starting confirm_batch_size"),
  First , put one less than the min batch size and expect that there are no
    Count = ?SOLRQ_BATCH_MIN_SETTING - 1,
    Count = put_objects(PBConn, BKey, Count),
    yz_rt:commit(Cluster, Index),
    verify_search_count(PBConn, Index, 0),

  Now , put one more and expect that all of them have been indexed ( because
  the solrq_batch_min has been reached ) and therefore show up in search
    1 = put_objects(PBConn, BKey, 1),
    yz_rt:commit(Cluster, Index),
    verify_search_count(PBConn, Index, ?SOLRQ_BATCH_MIN_SETTING),

  Finally , put another batch of one less than solrq_batch_min , but this
  have waited at least SOLRQ_DELAYMS_MAX ms .
    Count = put_objects(PBConn, BKey, Count),
    yz_rt:commit(Cluster, Index),
    verify_search_count(PBConn, Index, ?SOLRQ_BATCH_MIN_SETTING),
    StartMS = current_ms(),
    wait_until_search_count(PBConn, Index, ?SOLRQ_BATCH_MIN_SETTING + Count),
    Condition = gteq(current_ms() - StartMS, ?SOLRQ_DELAYMS_MAX),
    case Condition of
        false -> lager:error("Flush interval is less than ~pms", [?SOLRQ_DELAYMS_MAX]);
        true -> ok
    end,
    ?assertEqual(Condition, true),
    lager:info("confirm_batch_size ok"),
    ok.

confirm_hwm(Cluster, PBConn, Bucket, Index, HWM) ->
    lager:info("Starting confirm_hwm"),
    yz_rt:drain_solrqs(Cluster),
    {OldMin, OldMax, OldDelay} = set_index(Cluster, Index, 1, 100, 100),
    try
        yz_rt:load_intercept_code(Cluster),
        yz_rt:intercept_index_batch(Cluster, index_batch_returns_other_error),
        yz_rt:set_hwm(Cluster, HWM),
        yz_rt:set_purge_strategy(Cluster, ?PURGE_NONE),
        ?assertEqual(HWM, put_objects(PBConn, Bucket, HWM + 1))
    after
        yz_rt:intercept_index_batch(Cluster, index_batch_call_orig),
        yz_rt:set_purge_strategy(Cluster, ?PURGE_ONE),
        yz_rt:set_hwm(Cluster, 10000),
        yz_rt:set_index(Cluster, Index, OldMin, OldMax, OldDelay)
    end,
    lager:info("confirm_hwm ok for HWM ~p", [HWM]),
    ok.

-spec gteq(number(), number()) -> boolean().
gteq(A, B) -> A >= B.

confirm_draining(Cluster, PBConn, Bucket, Index) ->
    lager:info("Starting confirm_draining"),
    Count = ?SOLRQ_BATCH_MIN_SETTING - 1,
    Count = put_objects(PBConn, Bucket, Count),
    yz_rt:commit(Cluster, Index),
    verify_search_count(PBConn, Index, 0),
    yz_rt:drain_solrqs(Cluster),
    yz_rt:commit(Cluster, Index),
    verify_search_count(PBConn, Index, Count),
    lager:info("confirm_draining ok"),
    ok.

confirm_requeue_undelivered([Node|_] = Cluster, PBConn, BKey, Index) ->
    lager:info("Starting confirm_requeue_undelivered"),
    yz_rt:load_intercept_code(Node),
    yz_rt:intercept_index_batch(Node, index_batch_returns_other_error),

    Count = ?SOLRQ_BATCH_MIN_SETTING,
    Count = put_objects(PBConn, BKey, Count),
    yz_rt:commit(Cluster, Index),

    verify_search_count(PBConn, Index, 0),

    yz_rt:intercept_index_batch(Node, index_batch_call_orig),
    yz_rt:drain_solrqs(Node),
    verify_search_count(PBConn, Index, Count),
    lager:info("confirm_requeue_undelivered ok"),
    ok.

confirm_no_contenttype_data(Cluster, PBConn, BKey, Index) ->
    lager:info("Starting confirm_no_contenttype_data"),
    yz_rt:set_index(Cluster, Index, 1, 100, 100),
    Count = 1,
    Count = put_no_contenttype_objects(PBConn, BKey, Count),
    yz_rt:commit(Cluster, Index),
    verify_search_count(PBConn, Index, 1),
    lager:info("confirm_no_contenttype_data ok"),
    ok.

confirm_purge_strategy(Cluster, PBConn) ->
    lager:info("Starting confirm_purge_strategy"),
    confirm_purge_one_strategy(Cluster, PBConn,
                               {?BUCKET5, ?INDEX5}),
    confirm_purge_idx_strategy(Cluster, PBConn,
                               {?BUCKET7, ?INDEX7}),
    confirm_purge_none_strategy(Cluster, PBConn,
                                {?BUCKET11, ?INDEX11}),
    ok.

confirm_purge_one_strategy(Cluster, PBConn, Bucket1Index1) ->
    PurgeResults = do_purge(Cluster, PBConn, Bucket1Index1, ?PURGE_ONE),
    check_one_purged(PurgeResults),
    lager:info("confirm_purge_one_strategy ok"),
    ok.

check_one_purged({Index1Written, Index1SearchResults} = TestResults) ->
    Condition = first_purged(Index1Written, Index1SearchResults),
    case Condition of
        false ->
            lager:error("check_one_purged error: ~p", [TestResults]);
        _ -> ok
    end,
    ?assertEqual(Condition, true),
    ok.

confirm_purge_idx_strategy(Cluster, PBConn, Bucket1Index1) ->
    PurgeResults = do_purge(Cluster, PBConn, Bucket1Index1, ?PURGE_IDX),
    check_idx_purged(PurgeResults),
    lager:info("confirm_purge_idx_strategy ok"),
    ok.

check_idx_purged({[_K1, _K2, _K3, _K4, K5] = _Index1Written, Index1SearchResults} = TestResults) ->
  Note the second condition , because we wrote to Index1
  Otherwise , it was the second indexq , and nothing should
    Condition = equal(Index1SearchResults, [K5]),
    case Condition of
        false ->
            lager:error("check_idx_purged error: ~p", [TestResults]);
        _ -> ok
    end,
    ?assertEqual(Condition, true),
    ok.

confirm_purge_none_strategy(Cluster, PBConn, Bucket1Index1) ->
    PurgeResults = do_purge(Cluster, PBConn, Bucket1Index1,
                            ?PURGE_NONE),
    check_none_purged(PurgeResults),
    lager:info("confirm_purge_none_strategy ok"),
    ok.

check_none_purged({Index1Written, Index1SearchResults} = TestResults) ->
    Condition =
        equal(Index1Written, Index1SearchResults),
    case Condition of
        false ->
            lager:error("check_none_purged error: ~p", [TestResults]);
        _ -> ok
    end,
    ?assertEqual(Condition, true),
    ok.

equal(Written, Searched) ->
    lists:sort(Written) == lists:sort(Searched).

first_purged([_H|T] = _Written, Searched) ->
    lists:sort(T) == lists:sort(Searched).

  All the purge tests use this function to load up 2 indexqs
  up to the HWM .   We do this by setting an intercept so that
  the call to Solr fails for all of these calls , which will trigger
  Once the fuses for the indexqs are blown , we send one message
  using the first index .   This will trigger a purge , whose behavior
  We then restore the original Solr behavior by reverting the intercept ,
do_purge([Node|_] = Cluster, PBConn,
         {Bucket, Index},
         PurgeStrategy) ->
    yz_rt:set_purge_strategy(Cluster, PurgeStrategy),
    yz_rt:set_index(Cluster, Index, 1, 100, 99999),
    yz_rt:set_hwm(Cluster, 4),
    TargetPartition = 1096126227998177188652763624537212264741949407232,
    RepresentativeKeys = find_representative_bkeys(TargetPartition, Index, Bucket),
    Key1 = lists:nth(1, RepresentativeKeys),
    Key2 = lists:nth(2, RepresentativeKeys),
    Key3 = lists:nth(3, RepresentativeKeys),
    Key4 = lists:nth(4, RepresentativeKeys),
    Key5 = lists:nth(5, RepresentativeKeys),
    try
        yz_rt:load_intercept_code(Node),
        yz_rt:intercept_index_batch(Node, index_batch_returns_other_error),
        [Key1] = put_bkey_objects(PBConn, [Key1]),
        [Key2] = put_bkey_objects(PBConn, [Key2]),
        [Key3] = put_bkey_objects(PBConn, [Key3]),
        [Key4] = put_bkey_objects(PBConn, [Key4]),
        yz_rt:wait_until_fuses_blown(Node, TargetPartition, [Index]),
  to { TargetPartition , Index1 } , should be blown .
  Send one more message through the Indexq , which
        yz_rt:set_index(Cluster, Index, 1, 100, 99999),
        F = fun() ->
            [Key5] = put_bkey_objects(PBConn, [Key5])
        end,
        case PurgeStrategy of
            ?PURGE_NONE ->
                spawn(F);
            _ ->
                F()
        end
    after
        yz_rt:intercept_index_batch(Node, index_batch_call_orig),
        yz_rt:wait_until_fuses_reset(Node, TargetPartition, [Index]),
        yz_rt:drain_solrqs(Node),
        yz_rt:commit(Cluster, Index)
    end,
  The first list is the set of bkeys we wrote .
  The second list is the set that are available for search .
    KeysFound = search_bkeys(PBConn, Index),
    {[Key1, Key2, Key3, Key4, Key5], KeysFound}.

-spec search_bkeys(pid(), index_name()) -> [bkey()].
search_bkeys(PBConn, Index) ->
    {ok, {search_results, SearchResults, _Score, _NumFound}} =
        riakc_pb_socket:search(PBConn, Index, <<"*:*">>),
    lists:map(
        fun({_Index, Fields}) ->
                {{proplists:get_value(<<"_yz_rt">>, Fields),
                  proplists:get_value(<<"_yz_rb">>, Fields)},
                 proplists:get_value(<<"_yz_rk">>, Fields)}
        end,
        SearchResults).

-spec find_representative_bkeys(p(),
                                index_name(),
                                bucket()) -> [bkey()].
find_representative_bkeys(Partition, Index, Bucket) ->
    Solrq = yz_solrq:worker_regname(Index, Partition),
    Representatives = find_representatives(Index, Bucket),
    {ok, BKeys} = dict:find(Solrq, Representatives),
    BKeys.

-spec find_representatives(index_name(), bucket()) -> dict().
find_representatives(Index, Bucket) ->
    BKeys =
        lists:map(
          fun(I) ->
                  {Bucket, erlang:list_to_binary(erlang:integer_to_list(I))}
          end,
          lists:seq(1, 1000)),
    yz_rt:find_representatives(Index, BKeys, ?RING_SIZE).

-spec put_no_contenttype_objects(pid(), bucket(), non_neg_integer()) -> non_neg_integer().
put_no_contenttype_objects(PBConn, Bucket, Count) ->
    put_objects(PBConn, Bucket, Count, undefined).

-spec put_objects(pid(), bucket(), non_neg_integer()) -> non_neg_integer().
put_objects(PBConn, Bucket, Count) ->
    put_objects(PBConn, Bucket, Count, "text/plain").

-spec put_objects(pid(), bucket(), non_neg_integer(), string()|undefined) -> non_neg_integer().
put_objects(PBConn, Bucket, Count, ContentType) ->
    Key = <<"same_key_for_everyone">>,
    RandVals = [yz_rt:random_binary(16) || _ <- lists:seq(1, Count)],
    Objects = [case ContentType of
                   undefined ->
                       riakc_obj:new(Bucket, Key, Val);
                   _ ->
                       riakc_obj:new(Bucket, Key, Val, ContentType)
               end || Val <- RandVals],
    lager:info("Writing ~p objects to Riak...", [length(Objects)]),
    Results = [riakc_pb_socket:put(PBConn, Obj, [return_head, {timeout, 1000}]) || Obj <- Objects],
    NoWritten = length(lists:filter(fun({Result, _}) ->
                                        ok =:= Result
                                    end,
                                    Results)),
    lager:info("Wrote ~p objects to Riak...", [NoWritten]),
    NoWritten.

-spec put_bkey_objects(pid(), [bkey()]) -> [bkey()].
put_bkey_objects(PBConn, BKeys) ->
    BKeyObjects = [{{Bucket, Key},
                    riakc_obj:new(Bucket, Key, Key, "text/plain")}
                   || {Bucket, Key} <- BKeys],
    lager:info("Writing ~p bkeys ~p to Riak...", [length(BKeys), BKeys]),
    Results = [{BKey, riakc_pb_socket:put(PBConn, Obj,
                                          [return_head, {timeout, 1000}])}
               || {BKey, Obj} <- BKeyObjects],
    lists:map(
      fun({BKey, {_Result, _}}) -> BKey end,
      lists:filter(
          fun({BKey, {Result, _}}) ->
              case Result of
                  ok -> true;
                  _ ->
                      lager:info("The result of a put was not ok.  BKey: ~p Result: ~p", [BKey, Result]),
                      false
              end
          end,
          Results)).

verify_search_count(PBConn, Index, Count) ->
    {ok, {search_results, _R, _Score, Found}} =
        riakc_pb_socket:search(PBConn, Index, <<"*:*">>),
    ?assertEqual(Count, Found).

wait_until_search_count(PBConn, Index, Count) ->
    F = fun() ->
        {ok, {search_results, _R, _Score, Found}} =
            riakc_pb_socket:search(PBConn, Index, <<"*:*">>),
        Found =:= Count
        end,
    rt:wait_until(F).

current_ms() ->
    {MegaSecs, Secs, MicroSecs} = os:timestamp(),
    (MegaSecs * 1000000 + Secs) * 1000 + round(MicroSecs / 1000).

set_index(Cluster, Index, Min, Max, Delay) ->
    {[[{_WorkerId, {ok, {OldMin, OldMax, OldDelay}}}|_]], _} =
        yz_rt:set_index(Cluster, Index, Min, Max, Delay),
    {OldMin, OldMax, OldDelay}.

