You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by yo...@apache.org on 2011/11/29 04:56:25 UTC
svn commit: r1207744 - in /lucene/dev/branches/solrcloud/solr:
core/src/java/org/apache/solr/update/ core/src/test/org/apache/solr/search/
test-framework/src/java/org/apache/solr/
Author: yonik
Date: Tue Nov 29 03:56:24 2011
New Revision: 1207744
URL: http://svn.apache.org/viewvc?rev=1207744&view=rev
Log:
SOLR-2808: buffered updates shouldn't affect version info
Modified:
lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/update/FSUpdateLog.java
lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/search/TestRealTimeGet.java
lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/search/TestRecovery.java
lucene/dev/branches/solrcloud/solr/test-framework/src/java/org/apache/solr/SolrTestCaseJ4.java
Modified: lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/update/FSUpdateLog.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/update/FSUpdateLog.java?rev=1207744&r1=1207743&r2=1207744&view=diff
==============================================================================
--- lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/update/FSUpdateLog.java (original)
+++ lucene/dev/branches/solrcloud/solr/core/src/java/org/apache/solr/update/FSUpdateLog.java Tue Nov 29 03:56:24 2011
@@ -229,7 +229,12 @@ public class FSUpdateLog extends UpdateL
ensureLog();
long pos = tlog.write(cmd);
LogPtr ptr = new LogPtr(pos, cmd.getVersion());
- map.put(cmd.getIndexedId(), ptr);
+
+ // only update our map if we're not buffering
+ if ((cmd.getFlags() & UpdateCommand.BUFFERING) == 0) {
+ map.put(cmd.getIndexedId(), ptr);
+ }
+
// SolrCore.verbose("TLOG: added id " + cmd.getPrintableId() + " to " + tlog + " " + ptr + " map=" + System.identityHashCode(map));
}
}
@@ -245,9 +250,14 @@ public class FSUpdateLog extends UpdateL
ensureLog();
long pos = tlog.writeDelete(cmd);
LogPtr ptr = new LogPtr(pos, cmd.version);
- map.put(br, ptr);
- oldDeletes.put(br, ptr);
+ // only update our map if we're not buffering
+ if ((cmd.getFlags() & UpdateCommand.BUFFERING) == 0) {
+ map.put(br, ptr);
+
+ oldDeletes.put(br, ptr);
+ }
+
// SolrCore.verbose("TLOG: added delete for id " + cmd.id + " to " + tlog + " " + ptr + " map=" + System.identityHashCode(map));
}
}
Modified: lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/search/TestRealTimeGet.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/search/TestRealTimeGet.java?rev=1207744&r1=1207743&r2=1207744&view=diff
==============================================================================
--- lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/search/TestRealTimeGet.java (original)
+++ lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/search/TestRealTimeGet.java Tue Nov 29 03:56:24 2011
@@ -383,7 +383,7 @@ public class TestRealTimeGet extends Sol
}
// assertU("<delete><id>" + id + "</id></delete>");
- Long version = deleteAndGetVersion(Integer.toString(id));
+ Long version = deleteAndGetVersion(Integer.toString(id), null);
model.put(id, new DocInfo(version, -nextVal));
if (VERBOSE) {
@@ -604,7 +604,7 @@ public class TestRealTimeGet extends Sol
if (oper < commitPercent + deletePercent) {
verbose("deleting id",id,"val=",nextVal);
- Long version = deleteAndGetVersion(Integer.toString(id));
+ Long version = deleteAndGetVersion(Integer.toString(id), null);
assertTrue(version < 0);
// only update model if the version is newer
@@ -969,32 +969,6 @@ public class TestRealTimeGet extends Sol
}
- private Long addAndGetVersion(SolrInputDocument sdoc, SolrParams params) throws Exception {
- String response = updateJ(jsonAdd(sdoc), params);
- Map rsp = (Map)ObjectBuilder.fromJSON(response);
- List lst = (List)rsp.get("adds");
- if (lst == null || lst.size() == 0) return null;
- return (Long) lst.get(1);
- }
-
- private Long deleteAndGetVersion(String id) throws Exception {
- String response = updateJ("{\"delete\":{\"id\":\""+id+"\"}}", null);
- Map rsp = (Map)ObjectBuilder.fromJSON(response);
- List lst = (List)rsp.get("deletes");
- if (lst == null || lst.size() == 0) return null;
- return (Long) lst.get(1);
- }
-
- private Long deleteAndGetVersion(String id, SolrParams params) throws Exception {
- String response = updateJ("{\"delete\":{\"id\":\""+id+"\"}}", params);
- Map rsp = (Map)ObjectBuilder.fromJSON(response);
- List lst = (List)rsp.get("deletes");
- if (lst == null || lst.size() == 0) return null;
- return (Long) lst.get(1);
- }
-
-
-
// The purpose of this test is to roughly model how solr uses lucene
IndexReader reader;
@Test
Modified: lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/search/TestRecovery.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/search/TestRecovery.java?rev=1207744&r1=1207743&r2=1207744&view=diff
==============================================================================
--- lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/search/TestRecovery.java (original)
+++ lucene/dev/branches/solrcloud/solr/core/src/test/org/apache/solr/search/TestRecovery.java Tue Nov 29 03:56:24 2011
@@ -17,6 +17,8 @@
package org.apache.solr.search;
+import org.apache.noggit.JSONUtil;
+import org.apache.noggit.ObjectBuilder;
import org.apache.solr.SolrTestCaseJ4;
import org.apache.solr.request.SolrQueryRequest;
import org.apache.solr.update.DirectUpdateHandler2;
@@ -27,11 +29,16 @@ import org.junit.BeforeClass;
import org.junit.Ignore;
import org.junit.Test;
+import java.util.List;
+import java.util.Map;
import java.util.concurrent.Future;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
+import static org.apache.solr.update.processor.DistributedUpdateProcessor.SEEN_LEADER;
+
public class TestRecovery extends SolrTestCaseJ4 {
+ private static String SEEN_LEADER_VAL="true"; // value that means we've seen the leader and have version info (i.e. we are a non-leader replica)
@BeforeClass
public static void beforeClass() throws Exception {
@@ -113,6 +120,7 @@ public class TestRecovery extends SolrTe
Thread.sleep(100);
assertEquals(permits, logReplay.availablePermits()); // no updates, so insure that recovery didn't run
+ assertEquals(UpdateLog.State.ACTIVE, h.getCore().getUpdateHandler().getUpdateLog().getState());
} finally {
FSUpdateLog.testing_logReplayHook = null;
@@ -123,38 +131,38 @@ public class TestRecovery extends SolrTe
@Test
public void testBuffering() throws Exception {
- try {
-
- DirectUpdateHandler2.commitOnClose = false;
- final Semaphore logReplay = new Semaphore(0);
- final Semaphore logReplayFinish = new Semaphore(0);
- FSUpdateLog.testing_logReplayHook = new Runnable() {
- @Override
- public void run() {
- try {
- logReplay.acquire();
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
+ DirectUpdateHandler2.commitOnClose = false;
+ final Semaphore logReplay = new Semaphore(0);
+ final Semaphore logReplayFinish = new Semaphore(0);
+
+ FSUpdateLog.testing_logReplayHook = new Runnable() {
+ @Override
+ public void run() {
+ try {
+ logReplay.acquire();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
}
- };
+ }
+ };
- FSUpdateLog.testing_logReplayFinishHook = new Runnable() {
- @Override
- public void run() {
- logReplayFinish.release();
- }
- };
+ FSUpdateLog.testing_logReplayFinishHook = new Runnable() {
+ @Override
+ public void run() {
+ logReplayFinish.release();
+ }
+ };
+ SolrQueryRequest req = req();
+ UpdateHandler uhandler = req.getCore().getUpdateHandler();
+ UpdateLog ulog = uhandler.getUpdateLog();
+
+ try {
clearIndex();
assertU(commit());
- SolrQueryRequest req = req();
- UpdateHandler uhandler = req.getCore().getUpdateHandler();
- UpdateLog ulog = uhandler.getUpdateLog();
-
assertEquals(UpdateLog.State.ACTIVE, ulog.getState());
ulog.bufferUpdates();
assertEquals(UpdateLog.State.BUFFERING, ulog.getState());
@@ -165,10 +173,12 @@ public class TestRecovery extends SolrTe
ulog.bufferUpdates();
assertEquals(UpdateLog.State.BUFFERING, ulog.getState());
- assertU(adoc("id","1"));
- assertU(adoc("id","2"));
- assertU(adoc("id","3"));
- assertU(delI("1"));
+ long version = 1;
+ // simulate updates from a leader
+ updateJ(jsonAdd(sdoc("id","1", "_version_",Long.toString(++version))), params(SEEN_LEADER,SEEN_LEADER_VAL));
+ updateJ(jsonAdd(sdoc("id","2", "_version_",Long.toString(++version))), params(SEEN_LEADER,SEEN_LEADER_VAL));
+ updateJ(jsonAdd(sdoc("id","3", "_version_",Long.toString(++version))), params(SEEN_LEADER,SEEN_LEADER_VAL));
+ deleteAndGetVersion("1", params(SEEN_LEADER,SEEN_LEADER_VAL, "_version_",Long.toString(++version)));
assertU(commit());
// updates should be buffered, so we should not see any results yet.
@@ -176,6 +186,14 @@ public class TestRecovery extends SolrTe
, "/response/numFound==0"
);
+ // real-time get should also not show anything (this could change in the future,
+ // but it's currently used for validating version numbers too, so it would
+ // be bad for updates to be visible if we're just buffering.
+ assertJQ(req("qt","/get", "id","3")
+ ,"=={'doc':null}"
+ );
+
+
rinfoFuture = ulog.applyBufferedUpdates();
assertTrue(rinfoFuture != null);
@@ -190,53 +208,47 @@ public class TestRecovery extends SolrTe
, "/response/numFound==2"
);
+ // move back to recovering
+ ulog.bufferUpdates();
+ assertEquals(UpdateLog.State.BUFFERING, ulog.getState());
+
+ // TODO
+ rinfoFuture = ulog.applyBufferedUpdates();
+ if (rinfoFuture != null) rinfoFuture.get();
+ assertEquals(UpdateLog.State.ACTIVE, ulog.getState()); // leave each test method in a good state
} finally {
FSUpdateLog.testing_logReplayHook = null;
FSUpdateLog.testing_logReplayFinishHook = null;
+ req().close();
}
}
-}
-
-/**
-
- - update processor directly log, or pass through to update handler?
- - only updates with versions (from leader) get logged?
- - could also log user updates and assign versions later - but couldn't tell them it if failed!
- - processor needs to know not to check versions
- - this suggests processor should just directly log.
-
- - we shouldn't be syncing when temporarily logging (this is called via run update processor currently, so we're ok)
-
- Transition from "recovering" to "active" - need to ensure that no updates are lost.
-
- - grab global write lock
- - change state to active
- - release global write lock
-
- - UpdateHandler.setState(BUFFER_UPDATES) BUFFER_UPDATES, REPLAY_BUFFERED_UPDATES, ACTIVE
-
- ulog.bufferUpdates
- ulog.replayBufferedUpdates
- ulog.makeActive
-
- recoverFrom
- bufferUpdates()
- replayUpdates()
- makeActive()
+ private static Long getVer(SolrQueryRequest req) throws Exception {
+ String response = JQ(req);
+ Map rsp = (Map) ObjectBuilder.fromJSON(response);
+ Map doc = null;
+ if (rsp.containsKey("doc")) {
+ doc = (Map)rsp.get("doc");
+ } else if (rsp.containsKey("docs")) {
+ List lst = (List)rsp.get("docs");
+ if (lst.size() > 0) {
+ doc = (Map)lst.get(0);
+ }
+ } else if (rsp.containsKey("response")) {
+ Map responseMap = (Map)rsp.get("response");
+ List lst = (List)responseMap.get("docs");
+ if (lst.size() > 0) {
+ doc = (Map)lst.get(0);
+ }
+ }
- bufferUpdates()
- replayBufferedUpdates() // block or provide a callback function when active?
- - TODO: what if there are failures while replaying buffered updates?
- - perhaps just provide functions to return stats about the last recovery?
- - number of updates buffered
- - number of updates errored
- - time?
+ if (doc == null) return null;
- -
+ return (Long)doc.get("_version_");
+ }
+}
-**/
\ No newline at end of file
Modified: lucene/dev/branches/solrcloud/solr/test-framework/src/java/org/apache/solr/SolrTestCaseJ4.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/solrcloud/solr/test-framework/src/java/org/apache/solr/SolrTestCaseJ4.java?rev=1207744&r1=1207743&r2=1207744&view=diff
==============================================================================
--- lucene/dev/branches/solrcloud/solr/test-framework/src/java/org/apache/solr/SolrTestCaseJ4.java (original)
+++ lucene/dev/branches/solrcloud/solr/test-framework/src/java/org/apache/solr/SolrTestCaseJ4.java Tue Nov 29 03:56:24 2011
@@ -407,6 +407,30 @@ public abstract class SolrTestCaseJ4 ext
}
}
+ /** Makes a query request and returns the JSON string response */
+ public static String JQ(SolrQueryRequest req) throws Exception {
+ SolrParams params = req.getParams();
+ if (!"json".equals(params.get("wt","xml")) || params.get("indent")==null) {
+ ModifiableSolrParams newParams = new ModifiableSolrParams(params);
+ newParams.set("wt","json");
+ if (params.get("indent")==null) newParams.set("indent","true");
+ req.setParams(newParams);
+ }
+
+ String response;
+ boolean failed=true;
+ try {
+ response = h.query(req);
+ failed = false;
+ } finally {
+ if (failed) {
+ log.error("REQUEST FAILED: " + req.getParamString());
+ }
+ }
+
+ return response;
+ }
+
/**
* Validates a query matches some JSON test expressions using the default double delta tollerance.
* @see JSONTestUtil#DEFAULT_DELTA
@@ -426,7 +450,7 @@ public abstract class SolrTestCaseJ4 ext
* matching more than what you want to test.
* </p>
* @param req Solr request to execute
- * @param delta tollerance allowed in comparing float/double values
+ * @param delta tolerance allowed in comparing float/double values
* @param tests JSON path expression + '==' + expected value
*/
public static void assertJQ(SolrQueryRequest req, double delta, String... tests) throws Exception {
@@ -769,6 +793,24 @@ public abstract class SolrTestCaseJ4 ext
return out.toString();
}
+
+ public static Long addAndGetVersion(SolrInputDocument sdoc, SolrParams params) throws Exception {
+ String response = updateJ(jsonAdd(sdoc), params);
+ Map rsp = (Map)ObjectBuilder.fromJSON(response);
+ List lst = (List)rsp.get("adds");
+ if (lst == null || lst.size() == 0) return null;
+ return (Long) lst.get(1);
+ }
+
+ public static Long deleteAndGetVersion(String id, SolrParams params) throws Exception {
+ String response = updateJ(jsonDelId(id), params);
+ Map rsp = (Map)ObjectBuilder.fromJSON(response);
+ List lst = (List)rsp.get("deletes");
+ if (lst == null || lst.size() == 0) return null;
+ return (Long) lst.get(1);
+ }
+
+
/////////////////////////////////////////////////////////////////////////////////////
//////////////////////////// random document / index creation ///////////////////////
/////////////////////////////////////////////////////////////////////////////////////