Merge pull request #576 from basho/merge/2.1-into-develop

Merge 2.1 into develop

Author: jvoegele

include/yokozuna.hrl

@@ -160,6 +160,10 @@
                         app_helper:get_env(riak_core, platform_data_dir)++"/yz")).
 -define(YZ_TEMP_DIR, app_helper:get_env(?YZ_APP_NAME, temp_dir,
                         app_helper:get_env(riak_core, platform_data_dir)++"/yz_temp")).
+%% The request timeout for Solr calls. Defaults to 60 seconds.
+-define(YZ_SOLR_REQUEST_TIMEOUT, app_helper:get_env(?YZ_APP_NAME,
+                                                    solr_request_timeout,
+                                                    60000)).
 -define(YZ_PRIV, code:priv_dir(?YZ_APP_NAME)).
 -define(YZ_CORE_CFG_FILE, "solrconfig.xml").
 -define(YZ_INDEX_CMD, #yz_index_cmd).
@@ -237,11 +241,12 @@
                       {partition, lp()} |
                       {limit, pos_integer()}].
 -type ed_continuation() :: none | base64().
+-type ed_pairs() :: [{DocID::binary(), Hash::base64()}].
 
 -record(entropy_data, {
           more=false :: boolean(),
           continuation :: ed_continuation(),
-          pairs :: [{DocID::binary(), Hash::base64()}]
+          pairs :: ed_pairs()
          }).
 -type entropy_data() :: #entropy_data{}.
 -type keydiff() :: hashtree:keydiff().
@@ -288,6 +293,8 @@
 -define(ERROR(Fmt), lager:error(Fmt)).
 -define(ERROR(Fmt, Args), lager:error(Fmt, Args)).
 -define(INFO(Fmt, Args), lager:info(Fmt, Args)).
+-define(NOTICE(Fmt, Args), lager:notice(Fmt, Args)).
+-define(NOTICE(Fmt), lager:notice(Fmt)).
 -define(WARN(Fmt, Args), lager:warning(Fmt, Args)).
 
 %%%===================================================================

java_src/com/basho/yokozuna/handler/EntropyData.java

@@ -18,7 +18,7 @@
 
 import java.io.IOException;
 
-import org.apache.commons.codec.binary.Base64;
+import javax.xml.bind.DatatypeConverter;
 
 import org.apache.lucene.util.Bits;
 import org.apache.lucene.util.BytesRef;
@@ -117,40 +117,41 @@ public void handleRequestBody(final SolrQueryRequest req, final SolrQueryRespons
                 tmp = te.next();
             }
 
-            String text;
-            String[] vals;
-            String docPartition;
-            String vsn;
-            String riakBType;
-            String riakBName;
-            String riakKey;
-            String hash;
             int count = 0;
             BytesRef current = null;
             final Bits liveDocs = rdr.getLiveDocs();
 
             while(!endOfItr(tmp) && count < n) {
                 if (isLive(liveDocs, te)) {
                     current = BytesRef.deepCopyOf(tmp);
-                    text = tmp.utf8ToString();
+                    final String text = tmp.utf8ToString();
                     if (log.isDebugEnabled()) {
                         log.debug("text: " + text);
                     }
-                    vals = text.split(" ");
+                    final String [] vals = text.split(" ");
 
-                    vsn = vals[0];
-                    docPartition = vals[1];
-                    riakBType = decodeBase64DocPart(vals[2]);
-                    riakBName = decodeBase64DocPart(vals[3]);
-                    riakKey = decodeBase64DocPart(vals[4]);
-                    hash = vals[5];
+                    final String docPartition = vals[1];
 
+                    /*
+                      If the partition matches the one we are looking for,
+                      parse the version, bkey, and object hash from the
+                      entropy data field (term).
+                    */
                     if (partition.equals(docPartition)) {
-                        SolrDocument tmpDoc = new SolrDocument();
+                        final String vsn = vals[0];
+
+                        final String [] decoded = decodeForVersion(vsn,
+                                                                   vals[2],
+                                                                   vals[3],
+                                                                   vals[4]);
+
+                        final String hash = vals[5];
+
+                        final SolrDocument tmpDoc = new SolrDocument();
                         tmpDoc.addField("vsn", vsn);
-                        tmpDoc.addField("riak_bucket_type", riakBType);
-                        tmpDoc.addField("riak_bucket_name", riakBName);
-                        tmpDoc.addField("riak_key", riakKey);
+                        tmpDoc.addField("riak_bucket_type", decoded[0]);
+                        tmpDoc.addField("riak_bucket_name", decoded[1]);
+                        tmpDoc.addField("riak_key", decoded[2]);
                         tmpDoc.addField("base64_hash", hash);
                         docs.add(tmpDoc);
                         count++;
@@ -163,7 +164,8 @@ public void handleRequestBody(final SolrQueryRequest req, final SolrQueryRespons
                 rsp.add("more", false);
             } else {
                 rsp.add("more", true);
-                final String newCont = Base64.encodeBase64URLSafeString(current.bytes);
+                final String newCont =
+                    org.apache.commons.codec.binary.Base64.encodeBase64URLSafeString(current.bytes);
                 // The continue context for next req to start where
                 // this one finished.
                 rsp.add("continuation", newCont);
@@ -182,7 +184,7 @@ static boolean isLive(final Bits liveDocs, final TermsEnum te) throws IOExceptio
     }
 
     static BytesRef decodeCont(final String cont) {
-        final byte[] bytes = Base64.decodeBase64(cont);
+        final byte[] bytes = org.apache.commons.codec.binary.Base64.decodeBase64(cont);
         return new BytesRef(bytes);
     }
 
@@ -209,12 +211,38 @@ public String getSource() {
         return "TODO: implement getSource";
     }
 
+    /**
+       @param vsn a String vsn number referring to the item's ed handler version
+       @param riakBType riak bucket-type
+       @param riakBName riak bucket-name
+       @param riakKey riak key
+       @return a String array consisting of a Bucket Type, Bucket Name, and Riak Key
+    */
+    private String [] decodeForVersion(String vsn, String riakBType, String riakBName, String riakKey) {
+        final String [] bKeyInfo;
+        switch(Integer.parseInt(vsn)) {
+            case 1:
+                bKeyInfo = new String [] {riakBType, riakBName, riakKey};
+                break;
+            default:
+                bKeyInfo = new String []
+                    {
+                        decodeBase64DocPart(riakBType),
+                        decodeBase64DocPart(riakBName),
+                        decodeBase64DocPart(riakKey)
+                    };
+                break;
+        }
+        return bKeyInfo;
+    }
+
     /**
        @param base64EncodedVal base64 encoded string
        @return a string of decoded base64 bytes
-     */
+    */
     private String decodeBase64DocPart(String base64EncodedVal) {
-        byte[] bytes = Base64.decodeBase64(base64EncodedVal);
-        return new String(bytes);
+        return new String(DatatypeConverter.parseBase64Binary(
+                              base64EncodedVal));
     }
 }
+

misc/bench/schemas/fruit_schema.xml

@@ -6,6 +6,8 @@
 
    <field name="text" type="text_ws" indexed="true" stored="false" multiValued="true"/>
 
+   <field name="date_register" type="tdate" indexed="true" stored="true" multiValued="false" />
+
    <field name="_version_" type="long" indexed="true" stored="true"/>
 
    <!-- Entropy Data: Data related to anti-entropy -->
@@ -42,6 +44,10 @@
 
     <fieldType name="long" class="solr.TrieLongField" precisionStep="0" positionIncrementGap="0"/>
 
+    <fieldType name="date" class="solr.TrieDateField" precisionStep="0" positionIncrementGap="0"/>
+    <!-- A Trie based date field for faster date range queries and date faceting. -->
+    <fieldType name="tdate" class="solr.TrieDateField" precisionStep="6" positionIncrementGap="0"/>
+
     <!-- A text field that only splits on whitespace for exact matching of words -->
     <fieldType name="text_ws" class="solr.TextField" positionIncrementGap="100">
       <analyzer>

priv/yokozuna.schema

@@ -59,3 +59,12 @@
   {datatype, directory},
   hidden
 ]}.
+
+%% @doc The timeout for ibrowse (ibrowse:send_req) requests to Solr endpoints.
+%% Defaults to 60 seconds. It will always round up to the nearest second, e.g.
+%% 1ms = 999 ms = 1s.
+{mapping, "search.solr.request_timeout", "yokozuna.solr_request_timeout", [
+  {default, "60s"},
+  {datatype, {duration, ms}},
+  hidden
+]}.

rebar.config

@@ -14,8 +14,8 @@
    {git, "git://github.com/etrepum/kvc.git", {tag, "v1.5.0"}}},
   {riak_kv, ".*",
    {git, "git://github.com/basho/riak_kv.git", {branch, "develop"}}},
-  {ibrowse, "4.0.2",
-   {git, "git://github.com/cmullaparthi/ibrowse.git", {tag, "v4.0.2"}}}
+  {ibrowse, "4.0.1",
+   {git, "git://github.com/cmullaparthi/ibrowse.git", {tag, "v4.0.1"}}}
  ]}.
 
 {pre_hooks, [{compile, "./tools/grab-solr.sh"}]}.

riak_test/aae_test.erl

@@ -105,6 +105,9 @@ aae_run(Cluster, Bucket, Index) ->
 
             verify_no_repair_for_non_indexed_data(Cluster, Bucket, PBConns),
 
+            verify_count_and_repair_after_error_value(Cluster, Bucket, Index,
+                                                         PBConns),
+
             yz_rt:close_pb_conns(PBConns),
 
             verify_no_repair_after_restart(Cluster),
@@ -321,6 +324,39 @@ verify_no_repair_for_non_indexed_data(Cluster, {BType, _Bucket}, PBConns) ->
 verify_no_repair_for_non_indexed_data(_Cluster, _Bucket, _PBConns) ->
     ok.
 
+-spec verify_count_and_repair_after_error_value([node()], bucket(),
+                                                index_name(), [pid()])
+                                               -> ok.
+verify_count_and_repair_after_error_value(Cluster, {BType, _Bucket}, Index,
+                                          PBConns) ->
+    lager:info("verify total count and repair occurred for failed-to-index (bad) data"),
+    Bucket = {BType, Index},
+
+    %% 1. write KV data to non-indexed bucket
+    Conn = yz_rt:select_random(PBConns),
+    lager:info("write 1 bad search field to bucket ~p", [Bucket]),
+    Obj = riakc_obj:new(Bucket, <<"akey_bad_data">>, <<"{\"date_register\":3333}">>,
+                        "application/json"),
+
+    ok = riakc_pb_socket:put(Conn, Obj),
+
+    %% 2. setup tracing to count repair calls
+    ok = yz_rt:count_calls(Cluster, ?REPAIR_MFA),
+
+    %% 3. wait for full exchange round
+    ok = yz_rt:wait_for_full_exchange_round(Cluster, now()),
+    ok = yz_rt:stop_tracing(),
+
+    %% 4. verify repair count is 0
+    ?assertEqual(?N * 1, yz_rt:get_call_count(Cluster, ?REPAIR_MFA)),
+
+    %% 5. verify count after expiration
+    verify_exchange_after_expire(Cluster, Index),
+
+    ok;
+verify_count_and_repair_after_error_value(_Cluster, _Bucket, _Index, _PBConns) ->
+    ok.
+
 verify_num_match(Cluster, Index, Num) ->
     verify_num_match(yokozuna, Cluster, Index, Num).
 
@@ -331,10 +367,10 @@ verify_num_match(Type, Cluster, Index, Num) ->
             if Type =:= solr ->
                     Shards = [{N, yz_rt:node_solr_port(N)} || N <- Cluster],
                     yz_rt:search_expect(Type, {Host, yz_rt:node_solr_port(Node)},
-                                        Index, "text", "apricot", Shards, Num);
+                                        Index, "*", "*", Shards, Num);
                true ->
                     yz_rt:search_expect(Type, HP,
-                                        Index, "text", "apricot", Num)
+                                        Index, "*", "*", Num)
             end
         end,
     yz_rt:wait_until(Cluster, F).

riak_test/intercepts/yz_solr_intercepts.erl

@@ -1,6 +1,25 @@
 -module(yz_solr_intercepts).
 -compile(export_all).
 
+-type index_name() :: binary().
+
+-define(FMT(S, Args), lists:flatten(io_lib:format(S, Args))).
+
+-spec slow_cores() -> {ok, []}.
 slow_cores() ->
     timer:sleep(6000),
     {ok, []}.
+
+-spec entropy_data_cant_complete(index_name(), list()) -> {error, term()}.
+entropy_data_cant_complete(Core, Filter) ->
+    Params = [{wt, json}|Filter] -- [{continuation, none}],
+    Params2 = proplists:substitute_aliases([{continuation, continue},
+                                            {limit,n}], Params),
+    Opts = [{response_format, binary}],
+    URL = ?FMT("~s/~s/entropy_data?~s",
+               [yz_solr:base_url(), Core, mochiweb_util:urlencode(Params2)]),
+    case ibrowse:send_req(URL, [], get, [], Opts, 0) of
+        Error ->
+            {error, Error}
+    end.
+

riak_test/yokozuna_essential.erl

@@ -60,8 +60,7 @@ confirm() ->
             setup_indexing(Cluster, PBConns, YZBenchDir),
             verify_non_existent_index(Cluster, <<"froot">>),
             {0, _} = yz_rt:load_data(Cluster, ?BUCKET, YZBenchDir, ?NUM_KEYS),
-            %% wait for soft-commit
-            timer:sleep(1100),
+            yz_rt:commit(Cluster, ?INDEX),
             Ref = async_query(Cluster, YZBenchDir),
             %% Verify data exists before running join
             timer:sleep(30000),