You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by ma...@apache.org on 2021/01/12 22:37:58 UTC
[lucene-solr] branch reference_impl_dev updated (77fccf3 -> acf5e99)
This is an automated email from the ASF dual-hosted git repository.
markrmiller pushed a change to branch reference_impl_dev
in repository https://gitbox.apache.org/repos/asf/lucene-solr.git.
from 77fccf3 @1250 Cleanup.
new 388f716 @1251 Cleanup some of the managed schema dangling threads.
new 5c418c8 @1252 Cleanup some of the index fetch for recovery dangling threads.
new acf5e99 @1253 Cleanup around hardening test issues.
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../apache/lucene/store/NRTCachingDirectory.java | 4 +-
.../java/org/apache/solr/cloud/LeaderElector.java | 8 +-
.../solr/cloud/ShardLeaderElectionContext.java | 2 +-
.../java/org/apache/solr/cloud/SyncStrategy.java | 18 ++--
.../java/org/apache/solr/cloud/ZkController.java | 27 +++---
.../java/org/apache/solr/cloud/ZkShardTerms.java | 2 +-
.../java/org/apache/solr/core/CoreContainer.java | 24 +++---
.../src/java/org/apache/solr/core/SolrCore.java | 4 +-
.../java/org/apache/solr/handler/IndexFetcher.java | 98 +++++++++++++---------
.../org/apache/solr/handler/SchemaHandler.java | 2 +-
.../java/org/apache/solr/schema/IndexSchema.java | 42 +---------
.../org/apache/solr/schema/ManagedIndexSchema.java | 24 +++---
.../solr/schema/ManagedIndexSchemaFactory.java | 53 ++++++++----
.../apache/solr/schema/ZkIndexSchemaReader.java | 51 +++++------
.../apache/solr/update/DefaultSolrCoreState.java | 9 +-
.../AddSchemaFieldsUpdateProcessorFactory.java | 67 +++++++++------
.../src/test/org/apache/solr/CursorPagingTest.java | 1 +
.../cloud/TestTolerantUpdateProcessorCloud.java | 1 -
.../solr/handler/TestReplicationHandler.java | 9 +-
.../org/apache/solr/schema/SchemaWatcherTest.java | 3 +-
.../solr/security/BasicAuthOnSingleNodeTest.java | 1 +
.../solr/servlet/HttpSolrCallGetCoreTest.java | 3 +-
.../solr/client/solrj/impl/Http2SolrClient.java | 3 -
.../apache/solr/common/PerThreadExecService.java | 7 +-
.../apache/solr/common/cloud/ZkStateReader.java | 4 +-
.../src/java/org/apache/solr/SolrTestCase.java | 1 +
26 files changed, 241 insertions(+), 227 deletions(-)
[lucene-solr] 02/03: @1252 Cleanup some of the index fetch for
recovery dangling threads.
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
markrmiller pushed a commit to branch reference_impl_dev
in repository https://gitbox.apache.org/repos/asf/lucene-solr.git
commit 5c418c8aff14f2bea0a21eda72a720c4c7de4b30
Author: markrmiller@gmail.com <ma...@gmail.com>
AuthorDate: Mon Jan 4 10:06:14 2021 -0600
@1252 Cleanup some of the index fetch for recovery dangling threads.
---
.../apache/lucene/store/NRTCachingDirectory.java | 4 +-
.../java/org/apache/solr/cloud/LeaderElector.java | 8 +-
.../solr/cloud/ShardLeaderElectionContext.java | 2 +-
.../java/org/apache/solr/cloud/SyncStrategy.java | 18 ++--
.../java/org/apache/solr/cloud/ZkShardTerms.java | 2 +-
.../java/org/apache/solr/core/CoreContainer.java | 4 +-
.../src/java/org/apache/solr/core/SolrCore.java | 4 +-
.../java/org/apache/solr/handler/IndexFetcher.java | 98 +++++++++++++---------
.../org/apache/solr/handler/SchemaHandler.java | 2 +-
.../java/org/apache/solr/schema/IndexSchema.java | 42 +---------
.../org/apache/solr/schema/ManagedIndexSchema.java | 24 +++---
.../solr/schema/ManagedIndexSchemaFactory.java | 51 +++++++----
.../apache/solr/schema/ZkIndexSchemaReader.java | 26 +++---
.../apache/solr/update/DefaultSolrCoreState.java | 9 +-
.../AddSchemaFieldsUpdateProcessorFactory.java | 18 ++--
.../src/test/org/apache/solr/CursorPagingTest.java | 1 +
.../org/apache/solr/cloud/MoveReplicaTest.java | 2 +
.../cloud/TestTolerantUpdateProcessorCloud.java | 1 -
.../solr/handler/TestReplicationHandler.java | 9 +-
.../org/apache/solr/schema/SchemaWatcherTest.java | 3 +-
.../solr/client/solrj/impl/Http2SolrClient.java | 3 -
.../apache/solr/common/cloud/ZkStateReader.java | 4 +-
22 files changed, 172 insertions(+), 163 deletions(-)
diff --git a/lucene/core/src/java/org/apache/lucene/store/NRTCachingDirectory.java b/lucene/core/src/java/org/apache/lucene/store/NRTCachingDirectory.java
index 5150e12..ec7e5b9 100644
--- a/lucene/core/src/java/org/apache/lucene/store/NRTCachingDirectory.java
+++ b/lucene/core/src/java/org/apache/lucene/store/NRTCachingDirectory.java
@@ -174,9 +174,9 @@ public class NRTCachingDirectory extends FilterDirectory implements Accountable
for(String fileName : fileNames) {
unCache(fileName);
}
- if (Boolean.getBoolean("solr.nrtDirSync")) { // nocommit
+ //if (Boolean.getBoolean("solr.nrtDirSync")) { // nocommit
in.sync(fileNames);
- }
+ //}
}
@Override
diff --git a/solr/core/src/java/org/apache/solr/cloud/LeaderElector.java b/solr/core/src/java/org/apache/solr/cloud/LeaderElector.java
index ee413d0..3da2f7a 100644
--- a/solr/core/src/java/org/apache/solr/cloud/LeaderElector.java
+++ b/solr/core/src/java/org/apache/solr/cloud/LeaderElector.java
@@ -211,18 +211,20 @@ public class LeaderElector implements Closeable {
if (oldWatcher != null) {
IOUtils.closeQuietly(oldWatcher);
}
- if (context.leaderSeqPath == null) {
- throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, "Election has been cancelled");
- }
+
watcher = new ElectionWatcher(context.leaderSeqPath, watchedNode, context);
zkClient.exists(watchedNode, watcher);
state = WAITING_IN_ELECTION;
if (log.isDebugEnabled()) log.debug("Watching path {} to know if I could be the leader, my node is {}", watchedNode, context.leaderSeqPath);
+
+ log.info("Start recovery for core {}", context.leaderProps.getName());
try (SolrCore core = zkController.getCoreContainer().getCore(context.leaderProps.getName())) {
if (core != null) {
// if (!core.getSolrCoreState().isRecoverying()) {
core.getSolrCoreState().doRecovery(core);
// }
+ } else {
+ log.warn("No core found to start recovery with {}", context.leaderProps.getName());
}
}
return false;
diff --git a/solr/core/src/java/org/apache/solr/cloud/ShardLeaderElectionContext.java b/solr/core/src/java/org/apache/solr/cloud/ShardLeaderElectionContext.java
index 8d85470..f02e7d3 100644
--- a/solr/core/src/java/org/apache/solr/cloud/ShardLeaderElectionContext.java
+++ b/solr/core/src/java/org/apache/solr/cloud/ShardLeaderElectionContext.java
@@ -153,7 +153,7 @@ final class ShardLeaderElectionContext extends ShardLeaderElectionContextBase {
return;
}
result = syncStrategy.sync(zkController, core, leaderProps, weAreReplacement);
- log.warn("Sync strategy result {}", result);
+ log.info("Sync strategy sync result {}", result);
success = result.isSuccess();
if (!success) {
diff --git a/solr/core/src/java/org/apache/solr/cloud/SyncStrategy.java b/solr/core/src/java/org/apache/solr/cloud/SyncStrategy.java
index 09a7eaf..c6d331c 100644
--- a/solr/core/src/java/org/apache/solr/cloud/SyncStrategy.java
+++ b/solr/core/src/java/org/apache/solr/cloud/SyncStrategy.java
@@ -102,15 +102,15 @@ public class SyncStrategy implements Closeable {
}
try {
- if (success) {
- log.info("Sync Success - now sync replicas to me");
-
- syncToMe(zkController, collection, shardId, leaderProps, core.getCoreDescriptor(), core.getUpdateHandler().getUpdateLog().getNumRecordsToKeep());
-
- } else {
- log.info("Leader's attempt to sync with shard failed, moving to the next candidate");
- // lets see who seems ahead...
- }
+// if (success) {
+// log.info("Sync Success - now sync replicas to me");
+//
+// syncToMe(zkController, collection, shardId, leaderProps, core.getCoreDescriptor(), core.getUpdateHandler().getUpdateLog().getNumRecordsToKeep());
+//
+// } else {
+// log.info("Leader's attempt to sync with shard failed, moving to the next candidate");
+// // lets see who seems ahead...
+// }
} catch (Exception e) {
ParWork.propagateInterrupt(e);
diff --git a/solr/core/src/java/org/apache/solr/cloud/ZkShardTerms.java b/solr/core/src/java/org/apache/solr/cloud/ZkShardTerms.java
index b19777d..5851483 100644
--- a/solr/core/src/java/org/apache/solr/cloud/ZkShardTerms.java
+++ b/solr/core/src/java/org/apache/solr/cloud/ZkShardTerms.java
@@ -121,7 +121,7 @@ public class ZkShardTerms implements Closeable {
public void ensureTermsIsHigher(String leader, Set<String> replicasNeedingRecovery) throws KeeperException, InterruptedException {
if (log.isDebugEnabled()) log.debug("ensureTermsIsHigher leader={} replicasNeedingRecvoery={}", leader, replicasNeedingRecovery);
if (replicasNeedingRecovery.isEmpty()) return;
-
+ registerTerm(leader);
ShardTerms newTerms;
while( (newTerms = terms.get().increaseTerms(leader, replicasNeedingRecovery)) != null) {
if (forceSaveTerms(newTerms)) return;
diff --git a/solr/core/src/java/org/apache/solr/core/CoreContainer.java b/solr/core/src/java/org/apache/solr/core/CoreContainer.java
index 47fb57a..f010991 100644
--- a/solr/core/src/java/org/apache/solr/core/CoreContainer.java
+++ b/solr/core/src/java/org/apache/solr/core/CoreContainer.java
@@ -415,10 +415,10 @@ public class CoreContainer implements Closeable {
containerProperties.putAll(cfg.getSolrProperties());
- solrCoreLoadExecutor = new PerThreadExecService(ParWork.getRootSharedExecutor(), Math.max(16, Runtime.getRuntime().availableProcessors() / 2),
+ solrCoreLoadExecutor = new PerThreadExecService(ParWork.getRootSharedExecutor(), Math.max(16, Runtime.getRuntime().availableProcessors()),
false, false);
- solrCoreCloseExecutor = new PerThreadExecService(ParWork.getRootSharedExecutor(), Math.max(16, Runtime.getRuntime().availableProcessors() / 2),
+ solrCoreCloseExecutor = new PerThreadExecService(ParWork.getRootSharedExecutor(), Math.max(16, Runtime.getRuntime().availableProcessors()),
false, false);
}
diff --git a/solr/core/src/java/org/apache/solr/core/SolrCore.java b/solr/core/src/java/org/apache/solr/core/SolrCore.java
index a1b7c97..8d06233 100644
--- a/solr/core/src/java/org/apache/solr/core/SolrCore.java
+++ b/solr/core/src/java/org/apache/solr/core/SolrCore.java
@@ -173,7 +173,6 @@ import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -362,6 +361,9 @@ public final class SolrCore implements SolrInfoBean, Closeable {
if (this.schema == replacementSchema) {
return;
}
+
+ log.info("Set latest schema for core={} schema={}", getName(), replacementSchema);
+
this.schema = replacementSchema;
final SimilarityFactory similarityFactory = replacementSchema.getSimilarityFactory();
diff --git a/solr/core/src/java/org/apache/solr/handler/IndexFetcher.java b/solr/core/src/java/org/apache/solr/handler/IndexFetcher.java
index 31ff599..be65b25 100644
--- a/solr/core/src/java/org/apache/solr/handler/IndexFetcher.java
+++ b/solr/core/src/java/org/apache/solr/handler/IndexFetcher.java
@@ -589,35 +589,37 @@ public class IndexFetcher {
try {
+ // we have to be careful and do this after we know isFullCopyNeeded won't be flipped
+ if (!isFullCopyNeeded) {
+ solrCore.getUpdateHandler().getSolrCoreState().closeIndexWriter(solrCore, true);
+ }
+
log.info("Starting download (fullCopy={}) to {}", isFullCopyNeeded, tmpIndexDir);
- successfulInstall = false;
- long bytesDownloaded;
+ successfulInstall = true;
+ boolean downloadFailed = false;
+ long bytesDownloaded = 0;
try {
bytesDownloaded = downloadIndexFiles(isFullCopyNeeded, indexDir, tmpIndexDir, indexDirPath, tmpIndexDirPath, latestGeneration);
} catch (CheckSumFailException e) {
- isFullCopyNeeded = true;
- bytesDownloaded = downloadIndexFiles(isFullCopyNeeded, indexDir, tmpIndexDir, indexDirPath, tmpIndexDirPath, latestGeneration);
- }
-
- // we have to be careful and do this after we know isFullCopyNeeded won't be flipped
- if (!isFullCopyNeeded) {
- solrCore.getUpdateHandler().getSolrCoreState().closeIndexWriter(solrCore, true);
+ downloadFailed = true;
+ successfulInstall = false;
}
final long timeTakenSeconds = getReplicationTimeElapsed();
final Long bytesDownloadedPerSecond = (timeTakenSeconds != 0 ? Long.valueOf(bytesDownloaded / timeTakenSeconds) : null);
log.info("Total time taken for download (fullCopy={},bytesDownloaded={}) : {} secs ({} bytes/sec) to {}",
isFullCopyNeeded, bytesDownloaded, timeTakenSeconds, bytesDownloadedPerSecond, tmpIndexDir);
- successfulInstall = true;
+
Collection<Map<String,Object>> modifiedConfFiles = getModifiedConfFiles(confFilesToDownload);
- if (!modifiedConfFiles.isEmpty()) {
+ if (!modifiedConfFiles.isEmpty() && !downloadFailed) {
reloadCore = true;
downloadConfFiles(confFilesToDownload, latestGeneration);
if (isFullCopyNeeded && successfulInstall) {
successfulInstall = solrCore.modifyIndexProps(tmpIdxDirName);
if (successfulInstall) deleteTmpIdxDir = false;
} else {
- successfulInstall = moveIndexFiles(tmpIndexDir, indexDir);
+ terminateAndWaitFsyncService();
+ moveIndexFiles(tmpIndexDir, indexDir);
}
if (successfulInstall) {
if (isFullCopyNeeded) {
@@ -637,19 +639,21 @@ public class IndexFetcher {
successfulInstall);// write to a file time of replication and
// conf files.
}
- } else {
- terminateAndWaitFsyncService();
+ } else if (!downloadFailed) {
if (isFullCopyNeeded && successfulInstall) {
+ terminateAndWaitFsyncService();
successfulInstall = solrCore.modifyIndexProps(tmpIdxDirName);
if (!successfulInstall) {
log.error("Modify index props failed");
}
if (successfulInstall) deleteTmpIdxDir = false;
} else if (successfulInstall) {
- successfulInstall = moveIndexFiles(tmpIndexDir, indexDir);
+ terminateAndWaitFsyncService();
+ moveIndexFiles(tmpIndexDir, indexDir);
if (!successfulInstall) {
log.error("Move index files failed");
+ throw new SolrException(ErrorCode.SERVER_ERROR, "Move index files failed");
}
}
if (successfulInstall) {
@@ -660,7 +664,7 @@ public class IndexFetcher {
} finally {
solrCore.searchEnabled = true;
solrCore.indexEnabled = true;
- if (!isFullCopyNeeded && successfulInstall) {
+ if (!isFullCopyNeeded) {
solrCore.getUpdateHandler().getSolrCoreState().openIndexWriter(solrCore);
}
}
@@ -680,7 +684,12 @@ public class IndexFetcher {
if (indexDir != null) {
log.info("removing old index directory {}", indexDir);
solrCore.getDirectoryFactory().doneWithDirectory(indexDir);
- solrCore.getDirectoryFactory().remove(indexDir);
+ try {
+ solrCore.getDirectoryFactory().remove(indexDir);
+ } catch (IllegalArgumentException e) {
+ if (log.isDebugEnabled()) log.debug("Error removing directory in IndexFetcher", e);
+ // could already be removed
+ }
}
}
if (isFullCopyNeeded) {
@@ -690,9 +699,7 @@ public class IndexFetcher {
openNewSearcherAndUpdateCommitPoint();
}
- if (!isFullCopyNeeded && !forceReplication && !successfulInstall) {
- cleanup(solrCore, tmpIndexDir, indexDir, deleteTmpIdxDir, tmpTlogDir, successfulInstall);
- cleanupDone = true;
+ if (!isFullCopyNeeded && !forceReplication && !successfulInstall && !abort) {
// we try with a full copy of the index
log.warn(
"Replication attempt was not successful - trying a full index replication reloadCore={}",
@@ -711,9 +718,7 @@ public class IndexFetcher {
throw new SolrException(ErrorCode.SERVER_ERROR, "Index fetch failed : ", e);
}
} finally {
- if (!cleanupDone) {
- cleanup(solrCore, tmpIndexDir, indexDir, deleteTmpIdxDir, tmpTlogDir, successfulInstall);
- }
+ cleanup(solrCore, tmpIndexDir, indexDir, deleteTmpIdxDir, tmpTlogDir, successfulInstall);
}
}
@@ -735,11 +740,11 @@ public class IndexFetcher {
// this can happen on shutdown, a fetch may be running in a thread after DirectoryFactory is closed
log.warn("Could not log failed replication details", e);
}
- }
-
- if (core.getCoreContainer().isZooKeeperAware()) {
- // we only track replication success in SolrCloud mode
- core.getUpdateHandler().getSolrCoreState().setLastReplicateIndexSuccess(successfulInstall);
+ } else {
+ if (core.getCoreContainer().isZooKeeperAware()) {
+ // we only track replication success in SolrCloud mode
+ core.getUpdateHandler().getSolrCoreState().setLastReplicateIndexSuccess(successfulInstall);
+ }
}
} finally {
@@ -949,6 +954,7 @@ public class IndexFetcher {
waitSearcher[0].get();
} catch (InterruptedException | ExecutionException e) {
ParWork.propagateInterrupt(e);
+ throw new SolrException(ErrorCode.SERVER_ERROR, e);
}
}
commitPoint = searcher.get().getIndexReader().getIndexCommit();
@@ -1109,6 +1115,7 @@ public class IndexFetcher {
if (stop) {
throw new AlreadyClosedException();
}
+ log.info("Downloaded {}", tmpIndexDir, file.get(NAME));
filesDownloaded.add(Collections.unmodifiableMap(file));
} else {
if (log.isDebugEnabled()) {
@@ -1320,12 +1327,12 @@ public class IndexFetcher {
* <p/>
*/
private boolean moveAFile(Directory tmpIdxDir, Directory indexDir, String fname) {
- if (log.isDebugEnabled()) log.debug("Moving file: {}", fname);
boolean success = false;
try {
+ if (log.isDebugEnabled()) log.debug("Moving file: {} size={}", fname, tmpIdxDir.fileLength(fname));
if (slowFileExists(indexDir, fname)) {
log.warn("Cannot complete replication attempt because file already exists: {}", fname);
-
+
// we fail - we downloaded the files we need, if we can't move one in, we can't
// count on the correct index
return false;
@@ -1709,13 +1716,17 @@ public class IndexFetcher {
}
}
}
+ } catch (Exception e) {
+ log.error("Problem fetching file", e);
+ throw e;
} finally {
- cleanup();
+ cleanup(null);
//if cleanup succeeds . The file is downloaded fully. do an fsync
fsyncServiceFuture = fsyncService.submit(() -> {
try {
+ log.info("Sync and close fetched file", file);
file.sync();
- } catch (IOException e) {
+ } catch (Exception e) {
fsyncException = e;
}
});
@@ -1767,13 +1778,17 @@ public class IndexFetcher {
}
//if everything is fine, write down the packet to the file
file.write(buf, packetSize);
+
bytesDownloaded += packetSize;
// nocommit
log.info("Fetched and wrote {} bytes of file: {}", bytesDownloaded, fileName);
//errorCount is always set to zero after a successful packet
errorCount = 0;
- if (bytesDownloaded >= size)
+ if (bytesDownloaded >= size) {
return 0;
+ } else {
+ return 1;
+ }
}
} catch (CheckSumFailException e) {
throw e;
@@ -1819,8 +1834,9 @@ public class IndexFetcher {
/**
* cleanup everything
+ * @param ex exception if failed
*/
- private void cleanup() {
+ private void cleanup(Exception ex) {
try {
file.close();
} catch (Exception e) {/* no-op */
@@ -1835,10 +1851,12 @@ public class IndexFetcher {
log.error("Error deleting file: {}", this.saveAs, e);
}
//if the failure is due to a user abort it is returned normally else an exception is thrown
- if (!stop)
- throw new SolrException(SolrException.ErrorCode.SERVER_ERROR,
- "Unable to download " + fileName + " completely. Downloaded "
- + bytesDownloaded + "!=" + size);
+ SolrException exp = new SolrException(SolrException.ErrorCode.SERVER_ERROR, "Unable to download " + fileName + " completely. Downloaded " + bytesDownloaded + "!=" + size);
+ if (ex != null) {
+ ex.addSuppressed(exp);
+ } else {
+ throw exp;
+ }
}
}
@@ -2022,9 +2040,9 @@ public class IndexFetcher {
return masterUrl;
}
- private static final int MAX_RETRIES = 5;
+ private static final int MAX_RETRIES = 2;
- private static final int NO_CONTENT = 1;
+ private static final int NO_CONTENT = 0;
private static final int ERR = 2;
diff --git a/solr/core/src/java/org/apache/solr/handler/SchemaHandler.java b/solr/core/src/java/org/apache/solr/handler/SchemaHandler.java
index cf51595..3cde92d 100644
--- a/solr/core/src/java/org/apache/solr/handler/SchemaHandler.java
+++ b/solr/core/src/java/org/apache/solr/handler/SchemaHandler.java
@@ -170,7 +170,7 @@ public class SchemaHandler extends RequestHandlerBase implements SolrCoreAware,
log.info("REFRESHING SCHEMA (refreshIfBelowVersion={}, currentVersion={}) before returning version!"
, refreshIfBelowVersion, zkVersion);
ZkIndexSchemaReader zkIndexSchemaReader = managed.getManagedIndexSchemaFactory().getZkIndexSchemaReader();
- managed = (ManagedIndexSchema) zkIndexSchemaReader.updateSchema(null, -1);
+ managed = (ManagedIndexSchema) zkIndexSchemaReader.updateSchema(null, -1, req.getCore());
zkVersion = managed.getSchemaZkVersion();
}
}
diff --git a/solr/core/src/java/org/apache/solr/schema/IndexSchema.java b/solr/core/src/java/org/apache/solr/schema/IndexSchema.java
index cb8e23d..93796b9 100644
--- a/solr/core/src/java/org/apache/solr/schema/IndexSchema.java
+++ b/solr/core/src/java/org/apache/solr/schema/IndexSchema.java
@@ -79,7 +79,6 @@ import java.util.SortedMap;
import java.util.TreeMap;
import java.util.TreeSet;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Function;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
@@ -1642,8 +1641,6 @@ public class IndexSchema {
/**
* Copies this schema, adds the given field to the copy
- * Requires synchronizing on the object returned by
- * {@link #getSchemaUpdateLock()}.
*
* @param newField the SchemaField to add
* @param persist to persist the schema or not
@@ -1660,8 +1657,6 @@ public class IndexSchema {
/**
* Copies this schema, adds the given field to the copy
- * Requires synchronizing on the object returned by
- * {@link #getSchemaUpdateLock()}.
*
* @param newField the SchemaField to add
* @param copyFieldNames 0 or more names of targets to copy this field to. The targets must already exist.
@@ -1674,8 +1669,6 @@ public class IndexSchema {
/**
* Copies this schema, adds the given fields to the copy.
- * Requires synchronizing on the object returned by
- * {@link #getSchemaUpdateLock()}.
*
* @param newFields the SchemaFields to add
* @return a new IndexSchema based on this schema with newFields added
@@ -1687,8 +1680,6 @@ public class IndexSchema {
/**
* Copies this schema, adds the given fields to the copy
- * Requires synchronizing on the object returned by
- * {@link #getSchemaUpdateLock()}.
*
* @param newFields the SchemaFields to add
* @param copyFieldNames 0 or more names of targets to copy this field to. The target fields must already exist.
@@ -1708,8 +1699,6 @@ public class IndexSchema {
* <p>
* The schema will not be persisted.
* <p>
- * Requires synchronizing on the object returned by
- * {@link #getSchemaUpdateLock()}.
*
* @param names the names of the fields to delete
* @return a new IndexSchema based on this schema with the named fields deleted
@@ -1728,7 +1717,6 @@ public class IndexSchema {
* <p>
* The schema will not be persisted.
* <p>
- * Requires synchronizing on the object returned by {@link #getSchemaUpdateLock()}.
*
* @param fieldName The name of the field to be replaced
* @param replacementFieldType The field type of the replacement field
@@ -1744,7 +1732,6 @@ public class IndexSchema {
/**
* Copies this schema, adds the given dynamic fields to the copy,
* Requires synchronizing on the object returned by
- * {@link #getSchemaUpdateLock()}.
*
* @param newDynamicFields the SchemaFields to add
* @param copyFieldNames 0 or more names of targets to copy this field to. The target fields must already exist.
@@ -1767,7 +1754,6 @@ public class IndexSchema {
* The schema will not be persisted.
* <p>
* Requires synchronizing on the object returned by
- * {@link #getSchemaUpdateLock()}.
*
* @param fieldNamePatterns the names of the dynamic fields to delete
* @return a new IndexSchema based on this schema with the named dynamic fields deleted
@@ -1786,7 +1772,6 @@ public class IndexSchema {
* <p>
* The schema will not be persisted.
* <p>
- * Requires synchronizing on the object returned by {@link #getSchemaUpdateLock()}.
*
* @param fieldNamePattern The glob for the dynamic field to be replaced
* @param replacementFieldType The field type of the replacement dynamic field
@@ -1802,9 +1787,6 @@ public class IndexSchema {
/**
* Copies this schema and adds the new copy fields to the copy
- * Requires synchronizing on the object returned by
- * {@link #getSchemaUpdateLock()}.
- *
* @see #addCopyFields(String,Collection,int) to limit the number of copied characters.
*
* @param copyFields Key is the name of the source field name, value is a collection of target field names. Fields must exist.
@@ -1820,9 +1802,6 @@ public class IndexSchema {
/**
* Copies this schema and adds the new copy fields to the copy.
*
- * Requires synchronizing on the object returned by
- * {@link #getSchemaUpdateLock()}
- *
* @param source source field name
* @param destinations collection of target field names
* @param maxChars max number of characters to copy from the source to each
@@ -1841,9 +1820,6 @@ public class IndexSchema {
* <p>
* The schema will not be persisted.
* <p>
- * Requires synchronizing on the object returned by
- * {@link #getSchemaUpdateLock()}.
- *
* @param copyFields Key is the name of the source field name, value is a collection of target field names.
* Each corresponding copy field directives must exist.
* @return The new Schema with the copy fields deleted
@@ -1892,21 +1868,7 @@ public class IndexSchema {
}
/**
- * Returns the schema update lock that should be synchronized on
- * to update the schema. Only applicable to mutable schemas.
- *
- * @return the schema update lock object to synchronize on
- */
- public ReentrantLock getSchemaUpdateLock() {
- String msg = "This IndexSchema is not mutable.";
- log.error(msg);
- throw new SolrException(ErrorCode.SERVER_ERROR, msg);
- }
-
- /**
- * Copies this schema, adds the given field type to the copy,
- * Requires synchronizing on the object returned by
- * {@link #getSchemaUpdateLock()}.
+ * Copies this schema, adds the given field type to the copy
*
* @param fieldTypeList a list of FieldTypes to add
* @param persist to persist the schema or not
@@ -1924,7 +1886,6 @@ public class IndexSchema {
* <p>
* The schema will not be persisted.
* <p>
- * Requires synchronizing on the object returned by {@link #getSchemaUpdateLock()}.
*
* @param names the names of the field types to delete
* @return a new IndexSchema based on this schema with the named field types deleted
@@ -1943,7 +1904,6 @@ public class IndexSchema {
* <p>
* The schema will not be persisted.
* <p>
- * Requires synchronizing on the object returned by {@link #getSchemaUpdateLock()}.
*
* @param typeName The name of the field type to be replaced
* @param replacementClassName The class name of the replacement field type
diff --git a/solr/core/src/java/org/apache/solr/schema/ManagedIndexSchema.java b/solr/core/src/java/org/apache/solr/schema/ManagedIndexSchema.java
index 7bf51e4..bb4d3c1 100644
--- a/solr/core/src/java/org/apache/solr/schema/ManagedIndexSchema.java
+++ b/solr/core/src/java/org/apache/solr/schema/ManagedIndexSchema.java
@@ -77,7 +77,6 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.locks.ReentrantLock;
/** Solr-managed schema - non-user-editable, but can be mutable via internal and external REST API requests. */
public final class ManagedIndexSchema extends IndexSchema {
@@ -93,8 +92,6 @@ public final class ManagedIndexSchema extends IndexSchema {
volatile String managedSchemaResourceName;
volatile int schemaZkVersion;
-
- final ReentrantLock schemaUpdateLock;
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
@@ -111,9 +108,12 @@ public final class ManagedIndexSchema extends IndexSchema {
this.collection = collection;
this.managedSchemaResourceName = managedSchemaResourceName;
this.schemaZkVersion = schemaZkVersion;
- this.schemaUpdateLock = new ReentrantLock();
}
-
+
+
+ public IndexSchemaFactory getSchemaFactory() {
+ return managedIndexSchemaFactory;
+ }
/**
* Persist the schema to local storage or to ZooKeeper
@@ -187,18 +187,19 @@ public final class ManagedIndexSchema extends IndexSchema {
try {
zkClient.create(managedSchemaPath, data, CreateMode.PERSISTENT, true);
log.info("Created and persisted managed schema znode at {}", managedSchemaPath);
+ schemaZkVersion = 0;
} catch (KeeperException.NodeExistsException e) {
// This is okay - do nothing and fall through
}
- schemaZkVersion = 0;
} else {
try {
// Assumption: the path exists
- ver = schemaZkVersion;
+ ver = getSchemaZkVersion();
Stat managedSchemaStat = zkClient.setData(managedSchemaPath, data, ver, true);
log.info("Persisted managed schema version {} at {}", managedSchemaStat.getVersion(), managedSchemaPath);
+ schemaZkVersion = managedSchemaStat.getVersion();
} catch (KeeperException.BadVersionException e) {
// try again with latest schemaZkVersion value
Stat stat = zkClient.exists(managedSchemaPath, null, true);
@@ -206,7 +207,7 @@ public final class ManagedIndexSchema extends IndexSchema {
if (stat != null) {
found = stat.getVersion();
}
- log.info("Bad version when trying to persist schema using {} found {}", ver, found);
+ log.info("Bad version when trying to persist schema using {} found {} schema {}", ver, found, this);
schemaChangedInZk = true;
}
@@ -1385,8 +1386,8 @@ public final class ManagedIndexSchema extends IndexSchema {
this.isMutable = isMutable;
this.managedSchemaResourceName = managedSchemaResourceName;
this.schemaZkVersion = schemaZkVersion;
- this.schemaUpdateLock = new ReentrantLock();
this.collection = collection;
+ log.info("Copy to new ManagedIndexSchemaFactory with version {}", schemaZkVersion);
}
/**
@@ -1436,9 +1437,4 @@ public final class ManagedIndexSchema extends IndexSchema {
newSchema.decoders = decoders;
return newSchema;
}
-
- @Override
- public ReentrantLock getSchemaUpdateLock() {
- return schemaUpdateLock;
- }
}
diff --git a/solr/core/src/java/org/apache/solr/schema/ManagedIndexSchemaFactory.java b/solr/core/src/java/org/apache/solr/schema/ManagedIndexSchemaFactory.java
index 7fa89fb..fed467a 100644
--- a/solr/core/src/java/org/apache/solr/schema/ManagedIndexSchemaFactory.java
+++ b/solr/core/src/java/org/apache/solr/schema/ManagedIndexSchemaFactory.java
@@ -57,11 +57,11 @@ public class ManagedIndexSchemaFactory extends IndexSchemaFactory implements Sol
private volatile String managedSchemaResourceName = DEFAULT_MANAGED_SCHEMA_RESOURCE_NAME;
private volatile String collection;
- private CoreContainer cc;
+ private volatile CoreContainer cc;
private volatile SolrCore core;
- private ReentrantLock schemaUpdateLock = new ReentrantLock();
+ private final ReentrantLock schemaUpdateLock = new ReentrantLock();
public String getManagedSchemaResourceName() { return managedSchemaResourceName; }
private volatile SolrConfig config;
@@ -337,8 +337,27 @@ public class ManagedIndexSchemaFactory extends IndexSchemaFactory implements Sol
}
final ZkSolrResourceLoader zkLoader = (ZkSolrResourceLoader)loader;
final SolrZkClient zkClient = zkLoader.getZkClient();
+
+
+ // Rename the non-managed schema znode in ZooKeeper
+ final String nonManagedSchemaPath = zkLoader.getConfigSetZkPath() + "/" + resourceName;
+ final String upgradedSchemaPath = nonManagedSchemaPath + UPGRADED_SCHEMA_EXTENSION;
+ String managedSchemaPath = zkLoader.getConfigSetZkPath() + "/" + managedSchemaResourceName;
+
+ try {
+ if (zkClient.exists(managedSchemaPath)) {
+ return;
+ }
+ } catch (KeeperException e) {
+ throw new SolrException(ErrorCode.SERVER_ERROR, e);
+ } catch (InterruptedException e) {
+ throw new SolrException(ErrorCode.SERVER_ERROR, e);
+ }
+
final String lockPath = zkLoader.getConfigSetZkPath() + "/schemaUpgrade.lock";
boolean locked = false;
+ ReentrantLock lock = getSchemaUpdateLock();
+ lock.lock();
try {
try {
zkClient.makePath(lockPath, null, CreateMode.EPHEMERAL, null, true, true);
@@ -347,18 +366,16 @@ public class ManagedIndexSchemaFactory extends IndexSchemaFactory implements Sol
// some other node already started the upgrade, or an error occurred - bail out
return;
}
- schema.persistManagedSchemaToZooKeeper(true); // Only create, don't update it if it already exists
+ schema.persistManagedSchemaToZooKeeper(true); // Only create, don't update it if it already exists
// After successfully persisting the managed schema, rename the non-managed
// schema znode by appending UPGRADED_SCHEMA_EXTENSION to its name.
- // Rename the non-managed schema znode in ZooKeeper
- final String nonManagedSchemaPath = zkLoader.getConfigSetZkPath() + "/" + resourceName;
try {
if (zkClient.exists(nonManagedSchemaPath)) {
// First, copy the non-managed schema znode content to the upgraded schema znode
byte[] bytes = zkClient.getData(nonManagedSchemaPath, null, null);
- final String upgradedSchemaPath = nonManagedSchemaPath + UPGRADED_SCHEMA_EXTENSION;
+
zkClient.mkdir(upgradedSchemaPath);
Stat stat = zkClient.setData(upgradedSchemaPath, bytes, true);
// Then delete the non-managed schema znode
@@ -387,15 +404,19 @@ public class ManagedIndexSchemaFactory extends IndexSchemaFactory implements Sol
log.warn(msg, e); // Log as warning and suppress the exception
}
} finally {
- if (locked) {
- // unlock
- try {
- zkClient.delete(lockPath, -1);
- } catch (KeeperException.NoNodeException nne) {
- // ignore - someone else deleted it
- } catch (Exception e) {
- log.warn("Unable to delete schema upgrade lock file {}", lockPath, e);
+ try {
+ if (locked) {
+ // unlock
+ try {
+ zkClient.delete(lockPath, -1);
+ } catch (KeeperException.NoNodeException nne) {
+ // ignore - someone else deleted it
+ } catch (Exception e) {
+ log.warn("Unable to delete schema upgrade lock file {}", lockPath, e);
+ }
}
+ } finally {
+ lock.unlock();
}
}
}
@@ -429,8 +450,8 @@ public class ManagedIndexSchemaFactory extends IndexSchemaFactory implements Sol
}
public void setSchema(ManagedIndexSchema schema) {
- this.schema = schema;
core.setLatestSchema(schema);
+ this.schema = schema;
}
public boolean isMutable() {
diff --git a/solr/core/src/java/org/apache/solr/schema/ZkIndexSchemaReader.java b/solr/core/src/java/org/apache/solr/schema/ZkIndexSchemaReader.java
index 261f3df..a82f88f 100644
--- a/solr/core/src/java/org/apache/solr/schema/ZkIndexSchemaReader.java
+++ b/solr/core/src/java/org/apache/solr/schema/ZkIndexSchemaReader.java
@@ -17,7 +17,6 @@
package org.apache.solr.schema;
import org.apache.solr.cloud.ZkSolrResourceLoader;
-import org.apache.solr.common.AlreadyClosedException;
import org.apache.solr.common.cloud.OnReconnect;
import org.apache.solr.common.cloud.SolrZkClient;
import org.apache.solr.common.util.IOUtils;
@@ -81,9 +80,7 @@ public class ZkIndexSchemaReader implements OnReconnect {
}
});
- createSchemaWatcher();
-
- updateSchema(schemaWatcher, -1);
+ updateSchema(schemaWatcher, -1, solrCore);
solrCore.getCoreContainer().getZkController().addOnReconnectListener(this);
}
@@ -125,7 +122,7 @@ public class ZkIndexSchemaReader implements OnReconnect {
}
log.info("A schema change: {}, has occurred - updating schema from ZooKeeper ...", event);
try {
- schemaReader.updateSchema(this, -1);
+ schemaReader.updateSchema(this, -1, null);
} catch (Exception e) {
log.error("", e);
}
@@ -147,7 +144,7 @@ public class ZkIndexSchemaReader implements OnReconnect {
// }
// package visibility for test purposes
- public IndexSchema updateSchema(Watcher watcher, int version) throws KeeperException, InterruptedException {
+ public IndexSchema updateSchema(Watcher watcher, int version, SolrCore core) throws KeeperException, InterruptedException {
ManagedIndexSchema newSchema;
ReentrantLock lock = getSchemaUpdateLock();
lock.lock();
@@ -161,14 +158,14 @@ public class ZkIndexSchemaReader implements OnReconnect {
}
int existsVersion = exists.getVersion();
-
- int v = managedIndexSchemaFactory.getSchema().getSchemaZkVersion();
-
- if (version > -1) {
- v = version;
+ int v;
+ if (core != null) {
+ v = ((ManagedIndexSchema) core.getLatestSchema()).getSchemaZkVersion();
+ } else {
+ v = managedIndexSchemaFactory.getSchema().getSchemaZkVersion();
}
- log.info("Retrieved schema version {} from Zookeeper, existing={}", existsVersion, v);
+ log.info("Retrieved schema version {} from Zookeeper, existing={} schema={}", existsVersion, v, managedIndexSchemaFactory.getSchema());
if (v >= existsVersion) {
log.info("Old schema version {} is >= found version {}", v, existsVersion);
@@ -184,6 +181,9 @@ public class ZkIndexSchemaReader implements OnReconnect {
newSchema = new ManagedIndexSchema(managedIndexSchemaFactory, collection, managedIndexSchemaFactory.getConfig(), resourceName, inputSource, managedIndexSchemaFactory.isMutable(), resourceName,
stat.getVersion());
managedIndexSchemaFactory.setSchema(newSchema);
+ if (core != null) {
+ core.setLatestSchema(newSchema);
+ }
long stop = System.nanoTime();
log.info("Finished refreshing schema in {} ms", TimeUnit.MILLISECONDS.convert(stop - start, TimeUnit.NANOSECONDS));
@@ -204,7 +204,7 @@ public class ZkIndexSchemaReader implements OnReconnect {
public void command() {
try {
// force update now as the schema may have changed while our zk session was expired
- updateSchema(schemaWatcher, -1);
+ updateSchema(schemaWatcher, -1, null);
} catch (Exception exc) {
log.error("Failed to update managed-schema watcher after session expiration due to: {}", exc);
}
diff --git a/solr/core/src/java/org/apache/solr/update/DefaultSolrCoreState.java b/solr/core/src/java/org/apache/solr/update/DefaultSolrCoreState.java
index a70a003..a877eed 100644
--- a/solr/core/src/java/org/apache/solr/update/DefaultSolrCoreState.java
+++ b/solr/core/src/java/org/apache/solr/update/DefaultSolrCoreState.java
@@ -25,7 +25,6 @@ import org.apache.solr.common.AlreadyClosedException;
import org.apache.solr.common.ParWork;
import org.apache.solr.common.SolrException;
import org.apache.solr.common.SolrException.ErrorCode;
-import org.apache.solr.common.cloud.Replica;
import org.apache.solr.core.CoreContainer;
import org.apache.solr.core.CoreDescriptor;
import org.apache.solr.core.DirectoryFactory;
@@ -42,7 +41,6 @@ import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
@@ -317,16 +315,18 @@ public final class DefaultSolrCoreState extends SolrCoreState implements Recover
@Override
public void doRecovery(SolrCore core) {
+
log.info("Do recovery for core {}", core.getName());
CoreContainer corecontainer = core.getCoreContainer();
- CoreDescriptor coreDescriptor = core.getCoreDescriptor();
+
Runnable recoveryTask = () -> {
try {
if (SKIP_AUTO_RECOVERY) {
log.warn("Skipping recovery according to sys prop solrcloud.skip.autorecovery");
return;
}
-
+ CoreDescriptor coreDescriptor = core.getCoreDescriptor();
+ MDCLoggingContext.setCoreDescriptor(corecontainer, coreDescriptor);
if (log.isDebugEnabled()) log.debug("Going to create and run RecoveryStrategy");
// try {
@@ -405,6 +405,7 @@ public final class DefaultSolrCoreState extends SolrCoreState implements Recover
if (recoveryLock.isHeldByCurrentThread()) {
recoveryLock.unlock();
}
+ MDCLoggingContext.clear();
}
};
boolean success = false;
diff --git a/solr/core/src/java/org/apache/solr/update/processor/AddSchemaFieldsUpdateProcessorFactory.java b/solr/core/src/java/org/apache/solr/update/processor/AddSchemaFieldsUpdateProcessorFactory.java
index a6152f3..66b5df9 100644
--- a/solr/core/src/java/org/apache/solr/update/processor/AddSchemaFieldsUpdateProcessorFactory.java
+++ b/solr/core/src/java/org/apache/solr/update/processor/AddSchemaFieldsUpdateProcessorFactory.java
@@ -397,15 +397,19 @@ public class AddSchemaFieldsUpdateProcessorFactory extends UpdateRequestProcesso
// use the cmd's schema rather than the latest, because the schema
// can be updated during processing. Using the cmd's schema guarantees
// this will be detected and the cmd's schema updated.
- IndexSchema oldSchema = cmd.getReq().getSchema();;
+ IndexSchema oldSchema;
IndexSchema newSchema = null;
- for (; ; ) {
+ for (int cnt = 0; ; cnt++) {
+
List<SchemaField> newFields = new ArrayList<>();
// Group copyField defs per field and then per maxChar, to adapt to IndexSchema API
// build a selector each time through the loop b/c the schema we are
// processing may have changed
try {
- oldSchema = cmd.getReq().getSchema();
+
+ // use latest schema
+ oldSchema = cmd.getReq().getCore().getLatestSchema();
+
FieldNameSelector selector = buildSelector(oldSchema);
Map<String,List<SolrInputField>> unknownFields = new HashMap<>();
getUnknownFields(selector, doc, unknownFields);
@@ -462,6 +466,9 @@ public class AddSchemaFieldsUpdateProcessorFactory extends UpdateRequestProcesso
}
newSchema = oldSchema.addFields(newFields, Collections.emptyMap(), false);
+
+ log.info("Old schema version for request is {} version for latest on core is {} new schema version={}", ((ManagedIndexSchema) oldSchema).getSchemaZkVersion(), ((ManagedIndexSchema) core.getLatestSchema()).getSchemaZkVersion(), ((ManagedIndexSchema) newSchema).getSchemaZkVersion());
+
// Add copyFields
for (Map.Entry<String,Map<Integer,List<CopyFieldDef>>> entry : newCopyFields.entrySet()) {
String srcField = entry.getKey();
@@ -490,10 +497,11 @@ public class AddSchemaFieldsUpdateProcessorFactory extends UpdateRequestProcesso
log.error("At least one field to be added already exists in the schema - retrying.");
cmd.getReq().updateSchemaToLatest();
} catch (ManagedIndexSchema.SchemaChangedInZkException e) {
- log.info("Schema changed while processing request ...");
try {
- ((ManagedIndexSchema) newSchema).getManagedIndexSchemaFactory().getZkIndexSchemaReader().updateSchema(null, -1);
+ ((ManagedIndexSchema) cmd.getReq().getSchema()).getManagedIndexSchemaFactory().getZkIndexSchemaReader().updateSchema(null, -1, cmd.getReq().getCore());
cmd.getReq().updateSchemaToLatest();
+
+ log.info("Schema changed while processing request ... current latest version {} try={}", ((ManagedIndexSchema) cmd.getReq().getSchema()).getSchemaZkVersion(), cnt);
} catch (KeeperException.SessionExpiredException keeperException) {
throw new SolrException(SERVER_ERROR, keeperException);
} catch (Exception e1) {
diff --git a/solr/core/src/test/org/apache/solr/CursorPagingTest.java b/solr/core/src/test/org/apache/solr/CursorPagingTest.java
index 86caa30..7e7ddc0 100644
--- a/solr/core/src/test/org/apache/solr/CursorPagingTest.java
+++ b/solr/core/src/test/org/apache/solr/CursorPagingTest.java
@@ -49,6 +49,7 @@ import java.util.UUID;
/**
* Tests of deep paging using {@link CursorMark} and {@link CursorMarkParams#CURSOR_MARK_PARAM}.
*/
+// TODO bad seed? DCC82A1EDB76AEC
public class CursorPagingTest extends SolrTestCaseJ4 {
/** solrconfig.xml file name, shared with other cursor related tests */
diff --git a/solr/core/src/test/org/apache/solr/cloud/MoveReplicaTest.java b/solr/core/src/test/org/apache/solr/cloud/MoveReplicaTest.java
index 582a57b..89fed01 100644
--- a/solr/core/src/test/org/apache/solr/cloud/MoveReplicaTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/MoveReplicaTest.java
@@ -37,6 +37,7 @@ import org.apache.solr.util.IdUtils;
import org.junit.After;
import org.junit.Before;
import org.junit.BeforeClass;
+import org.junit.Ignore;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -50,6 +51,7 @@ import java.util.Map;
import java.util.Set;
@LuceneTestCase.SuppressCodecs({"MockRandom", "Direct", "SimpleText"})
+@Ignore // nocommit
public class MoveReplicaTest extends SolrCloudTestCase {
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
diff --git a/solr/core/src/test/org/apache/solr/cloud/TestTolerantUpdateProcessorCloud.java b/solr/core/src/test/org/apache/solr/cloud/TestTolerantUpdateProcessorCloud.java
index e2bcd46..a6a71b2 100644
--- a/solr/core/src/test/org/apache/solr/cloud/TestTolerantUpdateProcessorCloud.java
+++ b/solr/core/src/test/org/apache/solr/cloud/TestTolerantUpdateProcessorCloud.java
@@ -20,7 +20,6 @@ import org.apache.lucene.util.LuceneTestCase;
import org.apache.solr.SolrTestCaseJ4;
import org.apache.solr.client.solrj.SolrClient;
import org.apache.solr.client.solrj.embedded.JettySolrRunner;
-import org.apache.solr.client.solrj.impl.BaseHttpSolrClient;
import org.apache.solr.client.solrj.impl.CloudHttp2SolrClient;
import org.apache.solr.client.solrj.impl.Http2SolrClient;
import org.apache.solr.client.solrj.request.CollectionAdminRequest;
diff --git a/solr/core/src/test/org/apache/solr/handler/TestReplicationHandler.java b/solr/core/src/test/org/apache/solr/handler/TestReplicationHandler.java
index 4307b4c..a3fbd0f 100644
--- a/solr/core/src/test/org/apache/solr/handler/TestReplicationHandler.java
+++ b/solr/core/src/test/org/apache/solr/handler/TestReplicationHandler.java
@@ -87,6 +87,7 @@ import org.apache.solr.util.TimeOut;
import org.junit.After;
import org.junit.Before;
import org.junit.BeforeClass;
+import org.junit.Ignore;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -133,7 +134,7 @@ public class TestReplicationHandler extends SolrTestCaseJ4 {
super.setUp();
// System.setProperty("solr.directoryFactory", "solr.StandardDirectoryFactory");
// For manual testing only
- // useFactory(null); // force an FS factory.
+ useFactory(null); // force an FS factory.
master = new SolrInstance(createTempDir("solr-instance").toFile(), "master", null);
master.setUp();
masterJetty = createAndStartJetty(master);
@@ -291,7 +292,7 @@ public class TestReplicationHandler extends SolrTestCaseJ4 {
}
private Http2SolrClient adminClient(SolrClient client) {
- String adminUrl = ((HttpSolrClient)client).getBaseURL().replace("/collection1", "");
+ String adminUrl = ((Http2SolrClient)client).getBaseURL().replace("/collection1", "");
return getHttpSolrClient(adminUrl);
}
@@ -506,6 +507,7 @@ public class TestReplicationHandler extends SolrTestCaseJ4 {
}
@Test
+ @Ignore // nocommit
public void doTestIndexAndConfigReplication() throws Exception {
TestInjection.delayBeforeSlaveCommitRefresh = random().nextInt(10);
@@ -1041,6 +1043,7 @@ public class TestReplicationHandler extends SolrTestCaseJ4 {
}
@Test
+ @Ignore // nocommit
public void doTestRepeater() throws Exception {
// no polling
slave.setTestPort(masterJetty.getLocalPort());
@@ -1648,7 +1651,7 @@ public class TestReplicationHandler extends SolrTestCaseJ4 {
@Override
public void run() {
- final int totalDocs = TestUtil.nextInt(random(), 1, 10);
+ final int totalDocs = TestUtil.nextInt(LuceneTestCase.random(), 1, 10);
for (int i = 0; i < totalDocs; i++) {
try {
index(masterClient, "id", i + startId, "name", TestUtil.randomSimpleString(random(), 1000 , 5000));
diff --git a/solr/core/src/test/org/apache/solr/schema/SchemaWatcherTest.java b/solr/core/src/test/org/apache/solr/schema/SchemaWatcherTest.java
index 26fddd0..5f34c45 100644
--- a/solr/core/src/test/org/apache/solr/schema/SchemaWatcherTest.java
+++ b/solr/core/src/test/org/apache/solr/schema/SchemaWatcherTest.java
@@ -27,7 +27,6 @@ import org.junit.Test;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.verifyZeroInteractions;
public class SchemaWatcherTest {
@@ -45,7 +44,7 @@ public class SchemaWatcherTest {
@Test
public void testProcess() throws Exception {
schemaWatcher.process(new WatchedEvent(EventType.NodeDataChanged, KeeperState.SyncConnected, "/test"));
- verify(mockSchemaReader).updateSchema(schemaWatcher, -1);
+ verify(mockSchemaReader).updateSchema(schemaWatcher, -1, null);
}
}
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/Http2SolrClient.java b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/Http2SolrClient.java
index 4241f2f..6298773 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/Http2SolrClient.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/Http2SolrClient.java
@@ -1437,12 +1437,9 @@ public class Http2SolrClient extends SolrClient {
private static class MyInputStreamResponseListener extends InputStreamResponseListener {
private final AsyncListener<InputStream> asyncListener;
- private final HttpClient httpClient;
- private volatile InputStream stream;
public MyInputStreamResponseListener(HttpClient httpClient, AsyncListener<InputStream> asyncListener) {
this.asyncListener = asyncListener;
- this.httpClient = httpClient;
}
@Override
diff --git a/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java b/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java
index 186070c..78b860a 100644
--- a/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java
+++ b/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java
@@ -1515,10 +1515,10 @@ public class ZkStateReader implements SolrCloseable, Replica.NodeNameToBaseUrl {
replicasMap.put(r.getName(), newReplica);
}
}
- } else if (state != null) {
+ } else if (state != null && !properties.get(ZkStateReader.STATE_PROP).equals(state.toString())) {
if (log.isDebugEnabled()) log.debug("std state, set to {}", state);
properties.put(ZkStateReader.STATE_PROP, state.toString());
- if (state != Replica.State.ACTIVE && "true".equals(properties.get(LEADER_PROP))) {
+ if ("true".equals(properties.get(LEADER_PROP))) {
properties.remove(LEADER_PROP);
}
}
[lucene-solr] 03/03: @1253 Cleanup around hardening test issues.
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
markrmiller pushed a commit to branch reference_impl_dev
in repository https://gitbox.apache.org/repos/asf/lucene-solr.git
commit acf5e99eae91f1c5f3cf9c8a7fe73d8cfaeccbdd
Author: markrmiller@gmail.com <ma...@gmail.com>
AuthorDate: Tue Jan 12 16:37:24 2021 -0600
@1253 Cleanup around hardening test issues.
---
.../java/org/apache/solr/cloud/ZkController.java | 27 +++++++++++++---------
.../java/org/apache/solr/core/CoreContainer.java | 20 ++++++++--------
.../AddSchemaFieldsUpdateProcessorFactory.java | 17 ++++++++++----
.../src/test/org/apache/solr/CursorPagingTest.java | 2 +-
.../org/apache/solr/cloud/MoveReplicaTest.java | 2 --
.../solr/security/BasicAuthOnSingleNodeTest.java | 1 +
.../solr/servlet/HttpSolrCallGetCoreTest.java | 3 ++-
.../apache/solr/common/PerThreadExecService.java | 7 +++---
.../src/java/org/apache/solr/SolrTestCase.java | 1 +
9 files changed, 48 insertions(+), 32 deletions(-)
diff --git a/solr/core/src/java/org/apache/solr/cloud/ZkController.java b/solr/core/src/java/org/apache/solr/cloud/ZkController.java
index 18d824e..e1e9193 100644
--- a/solr/core/src/java/org/apache/solr/cloud/ZkController.java
+++ b/solr/core/src/java/org/apache/solr/cloud/ZkController.java
@@ -328,7 +328,7 @@ public class ZkController implements Closeable, Runnable {
log.info("Registering core {} afterExpiration? {}", descriptor.getName(), afterExpiration);
}
- if (zkController.isDcCalled() || zkController.getCoreContainer().isShutDown()) {
+ if (zkController.isDcCalled() || zkController.getCoreContainer().isShutDown() || (afterExpiration && !descriptor.getCloudDescriptor().hasRegistered())) {
return null;
}
zkController.register(descriptor.getName(), descriptor, afterExpiration);
@@ -1332,23 +1332,20 @@ public class ZkController implements Closeable, Runnable {
}
MDCLoggingContext.setCoreDescriptor(cc, desc);
ZkShardTerms shardTerms = null;
+ LeaderElector leaderElector = null;
try {
final String baseUrl = getBaseUrl();
final CloudDescriptor cloudDesc = desc.getCloudDescriptor();
final String collection = cloudDesc.getCollectionName();
final String shardId = cloudDesc.getShardId();
- log.info("Register terms for replica {}", coreName);
- ZkCollectionTerms ct = createCollectionTerms(collection);
- shardTerms = getShardTerms(collection, cloudDesc.getShardId());
-
+ log.info("Register SolrCore, core={} baseUrl={} collection={}, shard={} skipRecovery={}", coreName, baseUrl, collection, shardId);
+ AtomicReference<DocCollection> coll = new AtomicReference<>();
+ AtomicReference<Replica> replicaRef = new AtomicReference<>();
// the watcher is added to a set so multiple calls of this method will left only one watcher
getZkStateReader().registerCore(cloudDesc.getCollectionName());
- log.info("Register SolrCore, core={} baseUrl={} collection={}, shard={} skipRecovery={}", coreName, baseUrl, collection, shardId);
- AtomicReference<DocCollection> coll = new AtomicReference<>();
- AtomicReference<Replica> replicaRef = new AtomicReference<>();
try {
log.info("Waiting to see our entry in state.json {}", desc.getName());
zkStateReader.waitForState(collection, Integer.getInteger("solr.zkregister.leaderwait", 10000), TimeUnit.MILLISECONDS, (l, c) -> { // nocommit timeout
@@ -1377,11 +1374,16 @@ public class ZkController implements Closeable, Runnable {
}
log.info("Register replica - core:{} address:{} collection:{} shard:{} type={}", coreName, baseUrl, collection, shardId, replica.getType());
- if (isDcCalled() || isClosed) {
+ if (isDcCalled()) {
+ log.info("Disconnect already called, won't register");
throw new AlreadyClosedException();
}
- LeaderElector leaderElector = leaderElectors.get(replica.getName());
+ log.info("Register terms for replica {}", coreName);
+ //ZkCollectionTerms ct = createCollectionTerms(collection);
+ shardTerms = getShardTerms(collection, cloudDesc.getShardId());
+
+ leaderElector = leaderElectors.get(replica.getName());
if (leaderElector == null) {
ContextKey contextKey = new ContextKey(collection, coreName);
leaderElector = new LeaderElector(this, contextKey);
@@ -1394,7 +1396,7 @@ public class ZkController implements Closeable, Runnable {
// If we're a preferred leader, insert ourselves at the head of the queue
boolean joinAtHead = replica.getBool(SliceMutator.PREFERRED_LEADER_PROP, false);
if (replica.getType() != Type.PULL) {
- ct.register(cloudDesc.getShardId(), coreName);
+ //getCollectionTerms(collection).register(cloudDesc.getShardId(), coreName);
// nocommit review
joinElection(desc, joinAtHead);
}
@@ -1499,6 +1501,9 @@ public class ZkController implements Closeable, Runnable {
log.info("SolrCore Registered, core{} baseUrl={} collection={}, shard={}", coreName, baseUrl, collection, shardId);
return shardId;
} finally {
+ if (isDcCalled() || isClosed()) {
+ IOUtils.closeQuietly(leaderElector);
+ }
MDCLoggingContext.clear();
}
}
diff --git a/solr/core/src/java/org/apache/solr/core/CoreContainer.java b/solr/core/src/java/org/apache/solr/core/CoreContainer.java
index f010991..e2748ec 100644
--- a/solr/core/src/java/org/apache/solr/core/CoreContainer.java
+++ b/solr/core/src/java/org/apache/solr/core/CoreContainer.java
@@ -899,9 +899,9 @@ public class CoreContainer implements Closeable {
if (isZooKeeperAware() && !CloudUtil.checkIfValidCloudCore(this, cd)) {
continue;
}
- if (isZooKeeperAware()) {
- ParWork.getRootSharedExecutor().submit(new ZkController.RegisterCoreAsync(zkSys.zkController, cd, false));
- }
+// if (isZooKeeperAware()) {
+// ParWork.getRootSharedExecutor().submit(new ZkController.RegisterCoreAsync(zkSys.zkController, cd, false));
+// }
coreLoadFutures.add(solrCoreLoadExecutor.submit(() -> {
SolrCore core;
try {
@@ -909,17 +909,20 @@ public class CoreContainer implements Closeable {
zkSys.getZkController().throwErrorIfReplicaReplaced(cd);
}
core = createFromDescriptor(cd, false);
+
} finally {
solrCores.markCoreAsNotLoading(cd);
}
-
+ if (isZooKeeperAware()) {
+ new ZkController.RegisterCoreAsync(zkSys.zkController, cd, false).call();
+ }
return core;
}));
}
}
-
-
-
+ if (zkSys != null && zkSys.getZkController() != null) {
+ zkSys.getZkController().createEphemeralLiveNode();
+ }
} finally {
startedLoadingCores = true;
@@ -935,9 +938,6 @@ public class CoreContainer implements Closeable {
}
}
}
- if (zkSys != null && zkSys.getZkController() != null) {
- zkSys.getZkController().createEphemeralLiveNode();
- }
}
if (isZooKeeperAware()) {
diff --git a/solr/core/src/java/org/apache/solr/update/processor/AddSchemaFieldsUpdateProcessorFactory.java b/solr/core/src/java/org/apache/solr/update/processor/AddSchemaFieldsUpdateProcessorFactory.java
index 66b5df9..3053feb 100644
--- a/solr/core/src/java/org/apache/solr/update/processor/AddSchemaFieldsUpdateProcessorFactory.java
+++ b/solr/core/src/java/org/apache/solr/update/processor/AddSchemaFieldsUpdateProcessorFactory.java
@@ -392,6 +392,18 @@ public class AddSchemaFieldsUpdateProcessorFactory extends UpdateRequestProcesso
final String message = "This IndexSchema is not mutable.";
throw new SolrException(BAD_REQUEST, message);
}
+ updateSchemaIfNeeded(cmd);
+
+ try {
+ super.processAdd(cmd);
+ } catch (ManagedIndexSchema.UnknownFieldException e) {
+ log.info("Unknown field, retry ...");
+ updateSchemaIfNeeded(cmd);
+ super.processAdd(cmd);
+ }
+ }
+
+ private void updateSchemaIfNeeded(AddUpdateCommand cmd) {
final SolrInputDocument doc = cmd.getSolrInputDocument();
final SolrCore core = cmd.getReq().getCore();
// use the cmd's schema rather than the latest, because the schema
@@ -407,8 +419,7 @@ public class AddSchemaFieldsUpdateProcessorFactory extends UpdateRequestProcesso
// processing may have changed
try {
- // use latest schema
- oldSchema = cmd.getReq().getCore().getLatestSchema();
+ oldSchema = cmd.getReq().getSchema();
FieldNameSelector selector = buildSelector(oldSchema);
Map<String,List<SolrInputField>> unknownFields = new HashMap<>();
@@ -510,8 +521,6 @@ public class AddSchemaFieldsUpdateProcessorFactory extends UpdateRequestProcesso
}
}
}
-
- super.processAdd(cmd);
}
/**
diff --git a/solr/core/src/test/org/apache/solr/CursorPagingTest.java b/solr/core/src/test/org/apache/solr/CursorPagingTest.java
index 7e7ddc0..498b3c7 100644
--- a/solr/core/src/test/org/apache/solr/CursorPagingTest.java
+++ b/solr/core/src/test/org/apache/solr/CursorPagingTest.java
@@ -49,7 +49,7 @@ import java.util.UUID;
/**
* Tests of deep paging using {@link CursorMark} and {@link CursorMarkParams#CURSOR_MARK_PARAM}.
*/
-// TODO bad seed? DCC82A1EDB76AEC
+// TODO bad seed? DCC82A1EDB76AEC 9637DF7A121FD190
public class CursorPagingTest extends SolrTestCaseJ4 {
/** solrconfig.xml file name, shared with other cursor related tests */
diff --git a/solr/core/src/test/org/apache/solr/cloud/MoveReplicaTest.java b/solr/core/src/test/org/apache/solr/cloud/MoveReplicaTest.java
index 89fed01..582a57b 100644
--- a/solr/core/src/test/org/apache/solr/cloud/MoveReplicaTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/MoveReplicaTest.java
@@ -37,7 +37,6 @@ import org.apache.solr.util.IdUtils;
import org.junit.After;
import org.junit.Before;
import org.junit.BeforeClass;
-import org.junit.Ignore;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -51,7 +50,6 @@ import java.util.Map;
import java.util.Set;
@LuceneTestCase.SuppressCodecs({"MockRandom", "Direct", "SimpleText"})
-@Ignore // nocommit
public class MoveReplicaTest extends SolrCloudTestCase {
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
diff --git a/solr/core/src/test/org/apache/solr/security/BasicAuthOnSingleNodeTest.java b/solr/core/src/test/org/apache/solr/security/BasicAuthOnSingleNodeTest.java
index 33477b1..8a6d30c 100644
--- a/solr/core/src/test/org/apache/solr/security/BasicAuthOnSingleNodeTest.java
+++ b/solr/core/src/test/org/apache/solr/security/BasicAuthOnSingleNodeTest.java
@@ -56,6 +56,7 @@ public class BasicAuthOnSingleNodeTest extends SolrCloudAuthTestCase {
}
@Test
+ @Ignore // nocommit
public void basicTest() throws Exception {
try (Http2SolrClient client = new Http2SolrClient.Builder(cluster.getJettySolrRunner(0).getBaseUrl().toString())
.build()){
diff --git a/solr/core/src/test/org/apache/solr/servlet/HttpSolrCallGetCoreTest.java b/solr/core/src/test/org/apache/solr/servlet/HttpSolrCallGetCoreTest.java
index 2212bbf..76cc093 100644
--- a/solr/core/src/test/org/apache/solr/servlet/HttpSolrCallGetCoreTest.java
+++ b/solr/core/src/test/org/apache/solr/servlet/HttpSolrCallGetCoreTest.java
@@ -31,9 +31,11 @@ import org.apache.solr.cloud.SolrCloudTestCase;
import org.eclipse.jetty.server.Request;
import org.eclipse.jetty.server.Response;
import org.junit.BeforeClass;
+import org.junit.Ignore;
import org.junit.Test;
// commented 4-Sep-2018 @LuceneTestCase.BadApple(bugUrl="https://issues.apache.org/jira/browse/SOLR-12028") // 2-Aug-2018
+@Ignore // nocommit
public class HttpSolrCallGetCoreTest extends SolrCloudTestCase {
private static final String COLLECTION = "collection1";
private static final int NUM_SHARD = 3;
@@ -47,7 +49,6 @@ public class HttpSolrCallGetCoreTest extends SolrCloudTestCase {
CollectionAdminRequest
.createCollection(COLLECTION, "config", NUM_SHARD, REPLICA_FACTOR)
- .setMaxShardsPerNode(NUM_SHARD * REPLICA_FACTOR)
.process(cluster.getSolrClient());
}
diff --git a/solr/solrj/src/java/org/apache/solr/common/PerThreadExecService.java b/solr/solrj/src/java/org/apache/solr/common/PerThreadExecService.java
index 27c5e9f..f8ccd29 100644
--- a/solr/solrj/src/java/org/apache/solr/common/PerThreadExecService.java
+++ b/solr/solrj/src/java/org/apache/solr/common/PerThreadExecService.java
@@ -132,9 +132,10 @@ public class PerThreadExecService extends AbstractExecutorService {
@Override
public void execute(Runnable runnable) {
- if (shutdown) {
- throw new RejectedExecutionException();
- }
+// if (shutdown) {
+// throw new RejectedExecutionException();
+// }
+
running.incrementAndGet();
if (runnable instanceof ParWork.SolrFutureTask && !((ParWork.SolrFutureTask) runnable).isCallerThreadAllowed()) {
if (noCallerRunsAvailableLimit) {
diff --git a/solr/test-framework/src/java/org/apache/solr/SolrTestCase.java b/solr/test-framework/src/java/org/apache/solr/SolrTestCase.java
index eeac492..efa6f0c 100644
--- a/solr/test-framework/src/java/org/apache/solr/SolrTestCase.java
+++ b/solr/test-framework/src/java/org/apache/solr/SolrTestCase.java
@@ -282,6 +282,7 @@ public class SolrTestCase extends LuceneTestCase {
// System.setProperty("solr.zkstatewriter.throttle", "30");
System.setProperty("solr.stateworkqueue.throttle", "0");
+ System.setProperty("zkReaderGetLeaderRetryTimeoutMs", "300");
System.setProperty("solr.enablePublicKeyHandler", "false");
System.setProperty("solr.zkregister.leaderwait", "3000");
[lucene-solr] 01/03: @1251 Cleanup some of the managed schema
dangling threads.
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
markrmiller pushed a commit to branch reference_impl_dev
in repository https://gitbox.apache.org/repos/asf/lucene-solr.git
commit 388f7169fdb8b647787f11c7ceb29c080e549b66
Author: markrmiller@gmail.com <ma...@gmail.com>
AuthorDate: Sat Jan 2 21:28:01 2021 -0600
@1251 Cleanup some of the managed schema dangling threads.
---
.../solr/schema/ManagedIndexSchemaFactory.java | 4 +--
.../apache/solr/schema/ZkIndexSchemaReader.java | 31 +++++++----------
.../AddSchemaFieldsUpdateProcessorFactory.java | 40 ++++++++++------------
3 files changed, 32 insertions(+), 43 deletions(-)
diff --git a/solr/core/src/java/org/apache/solr/schema/ManagedIndexSchemaFactory.java b/solr/core/src/java/org/apache/solr/schema/ManagedIndexSchemaFactory.java
index fadd6e9..7fa89fb 100644
--- a/solr/core/src/java/org/apache/solr/schema/ManagedIndexSchemaFactory.java
+++ b/solr/core/src/java/org/apache/solr/schema/ManagedIndexSchemaFactory.java
@@ -61,6 +61,8 @@ public class ManagedIndexSchemaFactory extends IndexSchemaFactory implements Sol
private volatile SolrCore core;
+ private ReentrantLock schemaUpdateLock = new ReentrantLock();
+
public String getManagedSchemaResourceName() { return managedSchemaResourceName; }
private volatile SolrConfig config;
private volatile ResourceLoader loader;
@@ -397,8 +399,6 @@ public class ManagedIndexSchemaFactory extends IndexSchemaFactory implements Sol
}
}
}
-
- private ReentrantLock schemaUpdateLock = new ReentrantLock();
public ReentrantLock getSchemaUpdateLock() { return schemaUpdateLock; }
@Override
diff --git a/solr/core/src/java/org/apache/solr/schema/ZkIndexSchemaReader.java b/solr/core/src/java/org/apache/solr/schema/ZkIndexSchemaReader.java
index 7bb6c18..261f3df 100644
--- a/solr/core/src/java/org/apache/solr/schema/ZkIndexSchemaReader.java
+++ b/solr/core/src/java/org/apache/solr/schema/ZkIndexSchemaReader.java
@@ -42,7 +42,7 @@ import java.util.concurrent.locks.ReentrantLock;
/** Keeps a ManagedIndexSchema up-to-date when changes are made to the serialized managed schema in ZooKeeper */
public class ZkIndexSchemaReader implements OnReconnect {
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
- private volatile ManagedIndexSchemaFactory managedIndexSchemaFactory;
+ private final ManagedIndexSchemaFactory managedIndexSchemaFactory;
private volatile SolrZkClient zkClient;
private final String managedSchemaPath;
private final String uniqueCoreId; // used in equals impl to uniquely identify the core that we're dependent on
@@ -148,35 +148,32 @@ public class ZkIndexSchemaReader implements OnReconnect {
// package visibility for test purposes
public IndexSchema updateSchema(Watcher watcher, int version) throws KeeperException, InterruptedException {
- ManagedIndexSchema newSchema = null;
- ReentrantLock lock = null;
+ ManagedIndexSchema newSchema;
+ ReentrantLock lock = getSchemaUpdateLock();
+ lock.lock();
try {
- lock = getSchemaUpdateLock();
- lock.lock();
Stat stat = new Stat();
-
+ createSchemaWatcher();
Stat exists = zkClient.exists(managedSchemaPath, schemaWatcher, true);
if (exists == null) {
log.info("{} does not exist yet, watching ...}", managedSchemaPath);
return null;
- } else {
- createSchemaWatcher();
}
int existsVersion = exists.getVersion();
int v = managedIndexSchemaFactory.getSchema().getSchemaZkVersion();
+ if (version > -1) {
+ v = version;
+ }
+
log.info("Retrieved schema version {} from Zookeeper, existing={}", existsVersion, v);
if (v >= existsVersion) {
log.info("Old schema version {} is >= found version {}", v, existsVersion);
- exists = zkClient.exists(managedSchemaPath, this.schemaWatcher, true);
- if (v >= exists.getVersion()) {
- return null;
- } else {
- createSchemaWatcher();
- }
+
+ return null;
}
long start = System.nanoTime();
@@ -190,13 +187,11 @@ public class ZkIndexSchemaReader implements OnReconnect {
long stop = System.nanoTime();
log.info("Finished refreshing schema in {} ms", TimeUnit.MILLISECONDS.convert(stop - start, TimeUnit.NANOSECONDS));
- } catch (NullPointerException e) {
- throw new AlreadyClosedException();
} catch (Exception e) {
log.error("Exception updating schema", e);
return null;
} finally {
- if (lock != null) lock.unlock();
+ if (lock != null && lock.isHeldByCurrentThread()) lock.unlock();
}
return newSchema;
}
@@ -208,8 +203,6 @@ public class ZkIndexSchemaReader implements OnReconnect {
@Override
public void command() {
try {
- // setup a new watcher to get notified when the managed schema changes
- createSchemaWatcher();
// force update now as the schema may have changed while our zk session was expired
updateSchema(schemaWatcher, -1);
} catch (Exception exc) {
diff --git a/solr/core/src/java/org/apache/solr/update/processor/AddSchemaFieldsUpdateProcessorFactory.java b/solr/core/src/java/org/apache/solr/update/processor/AddSchemaFieldsUpdateProcessorFactory.java
index f3a7d45..a6152f3 100644
--- a/solr/core/src/java/org/apache/solr/update/processor/AddSchemaFieldsUpdateProcessorFactory.java
+++ b/solr/core/src/java/org/apache/solr/update/processor/AddSchemaFieldsUpdateProcessorFactory.java
@@ -24,6 +24,7 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;
import org.apache.solr.common.SolrException;
@@ -424,7 +425,7 @@ public class AddSchemaFieldsUpdateProcessorFactory extends UpdateRequestProcesso
}
if (newFields.isEmpty() && newCopyFields.isEmpty()) {
// nothing to do - no fields will be added - exit from the retry loop
- log.debug("No fields or copyFields to add to the schema.");
+ if (log.isDebugEnabled()) log.debug("No fields or copyFields to add to the schema.");
break;
} else if (isImmutableConfigSet(core)) {
final String message = "This ConfigSet is immutable.";
@@ -469,12 +470,18 @@ public class AddSchemaFieldsUpdateProcessorFactory extends UpdateRequestProcesso
}
}
if (null != newSchema) {
- boolean success = ((ManagedIndexSchema) newSchema).persistManagedSchema(false);
- if (log.isDebugEnabled()) log.debug("Successfully added field(s) and copyField(s) to the schema.");
- if (success) {
- core.setLatestSchema(newSchema);
- cmd.getReq().updateSchemaToLatest();
- break; // success - exit from the retry loop
+ ReentrantLock lock = ((ManagedIndexSchema) newSchema).getManagedIndexSchemaFactory().getSchemaUpdateLock();
+ lock.lock();
+ try {
+ boolean success = ((ManagedIndexSchema) newSchema).persistManagedSchema(false);
+ if (log.isDebugEnabled()) log.debug("Successfully added field(s) and copyField(s) to the schema.");
+ if (success) {
+ core.setLatestSchema(newSchema);
+ cmd.getReq().updateSchemaToLatest();
+ break; // success - exit from the retry loop
+ }
+ } finally {
+ lock.unlock();
}
} else {
throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, "Failed to add fields and/or copyFields.");
@@ -485,25 +492,14 @@ public class AddSchemaFieldsUpdateProcessorFactory extends UpdateRequestProcesso
} catch (ManagedIndexSchema.SchemaChangedInZkException e) {
log.info("Schema changed while processing request ...");
try {
- ((ManagedIndexSchema) newSchema).getManagedIndexSchemaFactory().getZkIndexSchemaReader().updateSchema(null, -1);
+ ((ManagedIndexSchema) newSchema).getManagedIndexSchemaFactory().getZkIndexSchemaReader().updateSchema(null, -1);
+ cmd.getReq().updateSchemaToLatest();
} catch (KeeperException.SessionExpiredException keeperException) {
throw new SolrException(SERVER_ERROR, keeperException);
} catch (Exception e1) {
- log.error("", e1);
+ log.error("Exception updating schema", e1);
+ throw new SolrException(SERVER_ERROR, e1);
}
-// if (newSchema != null) {
-//
-// cmd.getReq().updateSchemaToLatest(newSchema);
-// cmd.getReq().getCore().setLatestSchema(newSchema);
-// newSchema.postReadInform();
-// newSchema.refreshAnalyzers();
-// break;
-// } else {
- cmd.getReq().updateSchemaToLatest();
-// cmd.getReq().getSchema().refreshAnalyzers();
-// cmd.getReq().getSchema().postReadInform();
-
- // }
}
}