1818
1919import net .sf .ehcache .CacheManager ;
2020import net .sf .ehcache .Element ;
21+ import org .apache .lucene .document .Document ;
22+ import org .apache .lucene .index .IndexWriter ;
23+ import org .apache .lucene .store .AlreadyClosedException ;
2124import org .springframework .context .annotation .Profile ;
2225import org .springframework .stereotype .Component ;
2326import stroom .cache .AbstractCacheBean ;
2831import stroom .index .shared .Index ;
2932import stroom .index .shared .IndexService ;
3033import stroom .index .shared .IndexShard ;
34+ import stroom .index .shared .IndexShard .IndexShardStatus ;
3135import stroom .index .shared .IndexShardKey ;
3236import stroom .index .shared .IndexShardService ;
3337import stroom .jobsystem .server .JobTrackedSchedule ;
5963import java .util .concurrent .TimeUnit ;
6064import java .util .concurrent .atomic .AtomicBoolean ;
6165import java .util .concurrent .atomic .AtomicInteger ;
66+ import java .util .concurrent .locks .Lock ;
6267
6368/**
6469 * Pool API into open index shards.
6570 */
66- @ Component ("indexShardWriterCache " )
71+ @ Component ("indexShardManager " )
6772@ Profile (StroomSpringProfiles .PROD )
6873@ EntityEventHandler (type = Index .ENTITY_TYPE )
69- public class IndexShardWriterCacheImpl extends AbstractCacheBean <IndexShardKey , IndexShardWriter >
70- implements IndexShardWriterCache , EntityEvent .Handler {
71- private static final LambdaLogger LOGGER = LambdaLoggerFactory .getLogger (IndexShardWriterCacheImpl .class );
74+ public class IndexShardManagerImpl extends AbstractCacheBean <IndexShardKey , IndexShardWriter >
75+ implements IndexShardManager , Indexer , EntityEvent .Handler {
76+ private static final LambdaLogger LOGGER = LambdaLoggerFactory .getLogger (IndexShardManagerImpl .class );
7277 private static final int MAX_CACHE_ENTRIES = 1000000 ;
7378
7479 private final IndexService indexService ;
@@ -81,9 +86,11 @@ public class IndexShardWriterCacheImpl extends AbstractCacheBean<IndexShardKey,
8186 private final ConcurrentHashMap <IndexShard , IndexShardWriter > ownedWriters = new ConcurrentHashMap <>();
8287 private final AtomicBoolean deletingShards = new AtomicBoolean ();
8388
89+ private final StripedLock keyLocks = new StripedLock ();
90+
8491 @ Inject
85- IndexShardWriterCacheImpl (final CacheManager cacheManager , final StroomPropertyService stroomPropertyService ,
86- final IndexService indexService , final IndexShardService indexShardService , final NodeCache nodeCache , final TaskManager taskManager ) {
92+ IndexShardManagerImpl (final CacheManager cacheManager , final StroomPropertyService stroomPropertyService ,
93+ final IndexService indexService , final IndexShardService indexShardService , final NodeCache nodeCache , final TaskManager taskManager ) {
8794 super (cacheManager , "Index Shard Writer Cache" , MAX_CACHE_ENTRIES );
8895 this .stroomPropertyService = stroomPropertyService ;
8996 this .indexService = indexService ;
@@ -95,6 +102,83 @@ public class IndexShardWriterCacheImpl extends AbstractCacheBean<IndexShardKey,
95102 setMaxLiveTime (1 , TimeUnit .DAYS );
96103 }
97104
105+ @ Override
106+ public void addDocument (final IndexShardKey indexShardKey , final Document document ) {
107+ if (document != null ) {
108+
109+ IndexShardWriter indexShardWriter = getIndexShardWriter (indexShardKey );
110+
111+ // Try and add the document silently without locking.
112+ boolean success = false ;
113+ try {
114+ indexShardWriter .addDocument (document );
115+ success = true ;
116+ } catch (final Throwable t ) {
117+ LOGGER .trace (t ::getMessage , t );
118+ }
119+
120+ // Attempt a few more times under lock.
121+ for (int i = 0 ; !success && i < 100 ; i ++) {
122+ // If we failed then try under lock to make sure we get a new writer.
123+ final Lock lock = keyLocks .getLockForKey (indexShardKey );
124+ lock .lock ();
125+ try {
126+ // Ask the cache for the current one (it might have been changed by another thread) and try again.
127+ indexShardWriter = getIndexShardWriter (indexShardKey );
128+ success = addDocument (indexShardWriter , document );
129+
130+ if (!success ) {
131+ // Failed to add it so remove this object from the cache and try to get another one.
132+ remove (indexShardKey );
133+ }
134+
135+ } finally {
136+ lock .unlock ();
137+ }
138+ }
139+
140+ // One final try that will throw an index exception.
141+ if (!success ) {
142+ try {
143+ indexShardWriter = getIndexShardWriter (indexShardKey );
144+ indexShardWriter .addDocument (document );
145+ } catch (final IndexException e ) {
146+ throw e ;
147+ } catch (final Throwable e ) {
148+ throw new IndexException (e .getMessage (), e );
149+ }
150+ }
151+ }
152+ }
153+
154+ private boolean addDocument (final IndexShardWriter indexShardWriter , final Document document ) {
155+ boolean success = false ;
156+ try {
157+ indexShardWriter .addDocument (document );
158+ success = true ;
159+ } catch (final AlreadyClosedException | IndexException e ) {
160+ LOGGER .debug (e ::getMessage , e );
161+
162+ } catch (final Throwable t ) {
163+ LOGGER .error (t ::getMessage , t );
164+
165+ // Mark the shard as corrupt as this should be the
166+ // only reason we can't add a document.
167+ indexShardWriter .setStatus (IndexShardStatus .CORRUPT );
168+ }
169+
170+ return success ;
171+ }
172+
173+ private IndexShardWriter getIndexShardWriter (final IndexShardKey indexShardKey ) {
174+ final IndexShardWriter indexShardWriter = get (indexShardKey );
175+ if (indexShardWriter == null ) {
176+ throw new IndexException ("Unable to get writer for index '" + indexShardKey .getIndex ().getName ()
177+ + "'. Please check the index has active volumes." );
178+ }
179+ return indexShardWriter ;
180+ }
181+
98182 /**
99183 * Overrides method in simple pool. Will be called when an item is created
100184 * by the pool.
@@ -124,7 +208,7 @@ private IndexShardWriter getExistingWriter(final IndexShardKey key) {
124208 // Look for closed, non deleted, non full, non corrupt index shard
125209 // to add to
126210 if (indexShard .getIndex ().getId () == key .getIndex ().getId ()
127- && writer .getPartition ().equals (key .getPartition ()) && writer .isOkToReuse ()) {
211+ && writer .getPartition ().equals (key .getPartition ()) && IndexShardStatus . CLOSED . equals ( writer .getStatus ()) && ! writer . isFull ()) {
128212 try {
129213 // Open the writer.
130214 final boolean success = writer .open (false );
@@ -310,7 +394,7 @@ private void deleteLogicallyDeleted() {
310394 final Entry <IndexShard , IndexShardWriter > entry = iter .next ();
311395 final IndexShardWriter writer = entry .getValue ();
312396 try {
313- if (writer != null && writer .isDeleted ( )) {
397+ if (writer != null && IndexShardStatus . DELETED . equals ( writer .getStatus () )) {
314398 LOGGER .debug (() -> "deleteLogicallyDeleted() - Deleting index shard " + writer .getIndexShard ().getId ());
315399
316400 if (writer .deleteFromDisk ()) {
@@ -488,8 +572,12 @@ public void execute() {
488572 }
489573
490574 @ Override
491- public IndexShardWriter getWriter (final IndexShard indexShard ) {
492- return ownedWriters .get (indexShard );
575+ public IndexWriter getWriter (final IndexShard indexShard ) {
576+ final IndexShardWriter indexShardWriter = ownedWriters .get (indexShard );
577+ if (indexShardWriter != null ) {
578+ return indexShardWriter .getWriter ();
579+ }
580+ return null ;
493581 }
494582
495583 @ Override
0 commit comments