org.apache.lucene.index.ConcurrentMergeScheduler#org.apache.lucene.store.AlreadyClosedException源码实例Demo

@Override
protected boolean maybeFailEngine(String source, Exception e) {
    boolean shouldFail = super.maybeFailEngine(source, e);
    if (shouldFail) {
        return true;
    }
    // Check for AlreadyClosedException -- ACE is a very special
    // exception that should only be thrown in a tragic event. we pass on the checks to failOnTragicEvent which will
    // throw and AssertionError if the tragic event condition is not met.
    if (e instanceof AlreadyClosedException) {
        return failOnTragicEvent((AlreadyClosedException)e);
    } else if (e != null &&
            ((indexWriter.isOpen() == false && indexWriter.getTragicException() == e)
                    || (translog.isOpen() == false && translog.getTragicException() == e))) {
        // this spot on - we are handling the tragic event exception here so we have to fail the engine
        // right away
        failEngine(source, e);
        return true;
    }
    return false;
}
 

public CommonStats(IndexShard indexShard, CommonStatsFlags flags) {
    CommonStatsFlags.Flag[] setFlags = flags.getFlags();
    for (CommonStatsFlags.Flag flag : setFlags) {
        try {
            switch (flag) {
                case Docs:
                    docs = indexShard.docStats();
                    break;
                case Store:
                    store = indexShard.storeStats();
                    break;
                default:
                    throw new IllegalStateException("Unknown Flag: " + flag);
            }
        } catch (AlreadyClosedException e) {
            // shard is closed - no stats is fine
        }
    }
}
 

@Override
public void run() {
  int i = 0;
  while (i < 10000) {
    try {
      if (data.size() <= i) {
        sleep(1);
        continue;
      }
      final String key = "key" + i;
      final String val = "value" + i;
      final List<Document> documents = index.searchForDocuments(new TermQuery(new Term(key, val)), 10, new Sort(new SortField(key, SortField.Type.STRING)));
      if (documents.size() != 1) {
        throw new RuntimeException("Invalid number of matching documents for " + key + ", found " + documents);
      }
      ++i;
    } catch (IOException ioe) {
      error = ioe;
      break;
    } catch (InterruptedException e) {
    } catch (AlreadyClosedException ace) {
      error = ace;
      break;
    }
  }
}
 

private void swapSearcher(final Directory dir) throws IOException {
  /*
   * opening a searcher is possibly very expensive.
   * We rather close it again if the Autocompleter was closed during
   * this operation than block access to the current searcher while opening.
   */
  final IndexSearcher indexSearcher = createSearcher(dir);
  synchronized (searcherLock) {
    if(closed){
      indexSearcher.close();
      throw new AlreadyClosedException("Autocompleter has been closed");
    }
    if (searcher != null) {
      searcher.close();
    }
    // set the autocomplete index in the sync block - ensure consistency.
    searcher = indexSearcher;
    this.autoCompleteIndex = dir;
  }
}
 

private void reopenIndexWritersIfNecessary() throws Exception {
	if (!indexWriter.isOpen()) {
		synchronized (this) {
			if (!indexWriter.isOpen()) {
				this.indexWriter = this.indexSegmentInterface.getIndexWriter(segmentNumber);
				this.directoryReader = DirectoryReader.open(indexWriter, indexConfig.getIndexSettings().getApplyUncommittedDeletes(), false);
			}
		}
	}

	//TODO: is this a real use case?
	try {
		taxoWriter.getSize();
	}
	catch (AlreadyClosedException e) {
		synchronized (this) {
			this.taxoWriter = this.indexSegmentInterface.getTaxoWriter(segmentNumber);
			this.taxoReader = new DirectoryTaxonomyReader(taxoWriter);
		}
	}

}
 

public void testInvalidVersions2() throws IOException {
  Directory dir = newDirectory();
  IndexWriterConfig iwc = newIndexWriterConfig(new MockAnalyzer(random()));
  iwc.setCodec(TestUtil.alwaysPostingsFormat(new IDVersionPostingsFormat()));
  RandomIndexWriter w = new RandomIndexWriter(random(), dir, iwc, false);
  Document doc = new Document();
  // Long.MAX_VALUE:
  doc.add(new StringAndPayloadField("id", "id", new BytesRef(new byte[] {(byte)0x7f, (byte)0xff, (byte)0xff, (byte)0xff, (byte)0xff, (byte)0xff, (byte)0xff, (byte)0xff})));
  expectThrows(IllegalArgumentException.class, () -> {
    w.addDocument(doc);
    w.commit(false);
  });
  expectThrows(AlreadyClosedException.class, () -> {
    w.addDocument(doc);
  });

  dir.close();
}
 

/**
 * Expert: decreases the refCount of this IndexReader
 * instance.  If the refCount drops to 0, then this
 * reader is closed.  If an exception is hit, the refCount
 * is unchanged.
 *
 * @throws IOException in case an IOException occurs in  doClose()
 *
 * @see #incRef
 */
@SuppressWarnings("try")
public final void decRef() throws IOException {
  // only check refcount here (don't call ensureOpen()), so we can
  // still close the reader if it was made invalid by a child:
  if (refCount.get() <= 0) {
    throw new AlreadyClosedException("this IndexReader is closed");
  }
  
  final int rc = refCount.decrementAndGet();
  if (rc == 0) {
    closed = true;
    try (Closeable finalizer = this::reportCloseToParentReaders;
          Closeable finalizer1 = this::notifyReaderClosedListeners) {
      doClose();
    }
  } else if (rc < 0) {
    throw new IllegalStateException("too many decRef calls: refCount is " + rc + " after decrement");
  }
}
 

@Override
protected boolean maybeFailEngine(String source, Exception e) {
    boolean shouldFail = super.maybeFailEngine(source, e);
    if (shouldFail) {
        return true;
    }
    // Check for AlreadyClosedException -- ACE is a very special
    // exception that should only be thrown in a tragic event. we pass on the checks to failOnTragicEvent which will
    // throw and AssertionError if the tragic event condition is not met.
    if (e instanceof AlreadyClosedException) {
        return failOnTragicEvent((AlreadyClosedException)e);
    } else if (e != null &&
            ((indexWriter.isOpen() == false && indexWriter.getTragicException() == e)
                    || (translog.isOpen() == false && translog.getTragicException() == e))) {
        // this spot on - we are handling the tragic event exception here so we have to fail the engine
        // right away
        failEngine(source, e);
        return true;
    }
    return false;
}
 

public void testEnsureOpen() throws Exception {
  Directory dir = newDirectory();
  new IndexWriter(dir, new IndexWriterConfig(null)).close();
  SearcherManager sm = new SearcherManager(dir, null);
  IndexSearcher s = sm.acquire();
  sm.close();
  
  // this should succeed;
  sm.release(s);
  
  // this should fail
  expectThrows(AlreadyClosedException.class, () -> {
    sm.acquire();
  });
  
  // this should fail
  expectThrows(AlreadyClosedException.class, () -> {
    sm.maybeRefresh();
  });

  dir.close();
}
 

public void testAfterClose() throws Exception {
  Directory dir1 = getAssertNoDeletesDirectory(newDirectory());
  IndexWriter writer = new IndexWriter(dir1, newIndexWriterConfig(new MockAnalyzer(random())));

  // create the index
  createIndexNoClose(false, "test", writer);

  DirectoryReader r = writer.getReader();
  writer.close();

  TestUtil.checkIndex(dir1);

  // reader should remain usable even after IndexWriter is closed:
  assertEquals(100, r.numDocs());
  Query q = new TermQuery(new Term("indexname", "test"));
  IndexSearcher searcher = newSearcher(r);
  assertEquals(100, searcher.count(q));

  expectThrows(AlreadyClosedException.class, () -> {
    DirectoryReader.openIfChanged(r);
  });

  r.close();
  dir1.close();
}
 

public void testCloseInnerReader() throws Exception {
  Directory dir1 = getDir1(random());
  LeafReader ir1 = getOnlyLeafReader(DirectoryReader.open(dir1));
  
  // with overlapping
  ParallelLeafReader pr = new ParallelLeafReader(true,
   new LeafReader[] {ir1},
   new LeafReader[] {ir1});

  ir1.close();
  
  // should already be closed because inner reader is closed!
  expectThrows(AlreadyClosedException.class, () -> {
    pr.document(0);
  });
  
  // noop:
  pr.close();
  dir1.close();
}
 

public void testRegisterListenerOnClosedReader() throws IOException {
  Directory dir = newDirectory();
  IndexWriter w = new IndexWriter(dir, newIndexWriterConfig());
  w.addDocument(new Document());
  DirectoryReader r = DirectoryReader.open(w);
  w.close();

  // The reader is open, everything should work
  r.getReaderCacheHelper().addClosedListener(key -> {});
  r.leaves().get(0).reader().getReaderCacheHelper().addClosedListener(key -> {});
  r.leaves().get(0).reader().getCoreCacheHelper().addClosedListener(key -> {});

  // But now we close
  r.close();
  expectThrows(AlreadyClosedException.class, () -> r.getReaderCacheHelper().addClosedListener(key -> {}));
  expectThrows(AlreadyClosedException.class, () -> r.leaves().get(0).reader().getReaderCacheHelper().addClosedListener(key -> {}));
  expectThrows(AlreadyClosedException.class, () -> r.leaves().get(0).reader().getCoreCacheHelper().addClosedListener(key -> {}));

  dir.close();
}
 

@Override
void handle(Throwable t) {
  boolean report = true;

  if (t instanceof AlreadyClosedException || t instanceof MergePolicy.MergeAbortedException || t instanceof NullPointerException) {
    report = !didClose;
  } else if (t instanceof FileNotFoundException || t instanceof NoSuchFileException)  {
    report = !didClose;
  } else if (t instanceof IOException)  {
    Throwable t2 = t.getCause();
    if (t2 instanceof MergePolicy.MergeAbortedException) {
      report = !didClose;
    }
  }
  if (report) {
    t.printStackTrace(System.out);
    synchronized(failures) {
      failures.add(t);
    }
  }
}
 

/**
 * Closes the handler, releases resources and flushes the recent index changes
 * to persistent storage.
 */
@Override
public synchronized void close() {
  if (isClosed) {
    throw new AlreadyClosedException("Already closed");
  }
  isClosed = true;
  try {
    nrtManager.close();
    if (analyzer != null) {
      analyzer.close();
    }
    nrtManagerReopenThread.close();
    indexWriter.close();
  } catch (IOException ex) {
    LOG.log(Level.SEVERE, "Failed to close the Lucene index", ex);
  }
  LOG.info("Successfully closed the Lucene index...");
}
 

private boolean failOnTragicEvent(AlreadyClosedException ex) {
    final boolean engineFailed;
    // if we are already closed due to some tragic exception
    // we need to fail the engine. it might have already been failed before
    // but we are double-checking it's failed and closed
    if (indexWriter.isOpen() == false && indexWriter.getTragicException() != null) {
        final Exception tragicException;
        if (indexWriter.getTragicException() instanceof Exception) {
            tragicException = (Exception) indexWriter.getTragicException();
        } else {
            tragicException = new RuntimeException(indexWriter.getTragicException());
        }
        failEngine("already closed by tragic event on the index writer", tragicException);
        engineFailed = true;
    } else if (translog.isOpen() == false && translog.getTragicException() != null) {
        failEngine("already closed by tragic event on the translog", translog.getTragicException());
        engineFailed = true;
    } else if (failedEngine.get() == null && isClosed.get() == false) { // we are closed but the engine is not failed yet?
        // this smells like a bug - we only expect ACE if we are in a fatal case ie. either translog or IW is closed by
        // a tragic event or has closed itself. if that is not the case we are in a buggy state and raise an assertion error
        throw new AssertionError("Unexpected AlreadyClosedException", ex);
    } else {
        engineFailed = false;
    }
    return engineFailed;
}
 

private void updateIndexWriterSettings() {
    try {
        final LiveIndexWriterConfig iwc = indexWriter.getConfig();
        iwc.setRAMBufferSizeMB(engineConfig.getIndexingBufferSize().mbFrac());
        iwc.setUseCompoundFile(engineConfig.isCompoundOnFlush());
    } catch (AlreadyClosedException ex) {
        // ignore
    }
}
 

private void updateIndexWriterSettings() {
    try {
        final LiveIndexWriterConfig iwc = indexWriter.getConfig();
        iwc.setRAMBufferSizeMB(engineConfig.getIndexingBufferSize().mbFrac());
        iwc.setUseCompoundFile(engineConfig.isCompoundOnFlush());
    } catch (AlreadyClosedException ex) {
        // ignore
    }
}
 

@Override
protected boolean maybeFailEngine(String source, Throwable t) {
    boolean shouldFail = super.maybeFailEngine(source, t);
    if (shouldFail) {
        return true;
    }

    // Check for AlreadyClosedException
    if (t instanceof AlreadyClosedException) {
        // if we are already closed due to some tragic exception
        // we need to fail the engine. it might have already been failed before
        // but we are double-checking it's failed and closed
        if (indexWriter.isOpen() == false && indexWriter.getTragicException() != null) {
            failEngine("already closed by tragic event on the index writer", indexWriter.getTragicException());
        } else if (translog.isOpen() == false && translog.getTragicException() != null) {
            failEngine("already closed by tragic event on the translog", translog.getTragicException());
        }
        return true;
    } else if (t != null &&
            ((indexWriter.isOpen() == false && indexWriter.getTragicException() == t)
                    || (translog.isOpen() == false && translog.getTragicException() == t))) {
        // this spot on - we are handling the tragic event exception here so we have to fail the engine
        // right away
        failEngine(source, t);
        return true;
    }
    return false;
}
 

public StoreStats storeStats() {
    try {
        return store.stats();
    } catch (IOException e) {
        logger.debug("io exception when get store stats", e);
        throw new ElasticsearchException("io exception while building 'store stats'", e);
    } catch (AlreadyClosedException ex) {
        logger.debug("shard already closed exception", ex);
        return null; // already closed
    }
}
 

public ShardRowContext(IndexShard indexShard, ClusterService clusterService) {
    this(indexShard, null, clusterService, Suppliers.memoizeWithExpiration(() -> {
        try {
            StoreStats storeStats = indexShard.storeStats();
            return storeStats.getSizeInBytes();
        } catch (AlreadyClosedException e) {
            return 0L;
        }
    }, 10, TimeUnit.SECONDS));
}
 

public static boolean isShardNotAvailableException(Throwable t) {
    Throwable actual = ExceptionsHelper.unwrapCause(t);
    if (actual instanceof ShardNotFoundException ||
            actual instanceof IndexNotFoundException ||
            actual instanceof IllegalIndexShardStateException ||
            actual instanceof NoShardAvailableActionException ||
            actual instanceof UnavailableShardsException ||
            actual instanceof AlreadyClosedException) {
        return true;
    }
    return false;
}
 

@Test
public void testShardSizeExpressionWhenIndexShardHasBeenClosed() {
    IndexShard mock = mockIndexShard();
    when(mock.storeStats()).thenThrow(new AlreadyClosedException("shard already closed"));

    ShardReferenceResolver resolver = new ShardReferenceResolver(schemas, new ShardRowContext(mock, clusterService));
    Reference refInfo = refInfo("sys.shards.size", DataTypes.LONG, RowGranularity.SHARD);
    NestableInput<Long> shardSizeExpression = (NestableInput<Long>) resolver.getImplementation(refInfo);
    assertThat(shardSizeExpression.value(), is(0L));
}
 

/**
 * Tests whether or not the engine should be flushed periodically.
 * This test is based on the current size of the translog compared to the configured flush threshold size.
 *
 * @return {@code true} if the engine should be flushed
 */
boolean shouldPeriodicallyFlush() {
    final Engine engine = getEngineOrNull();
    if (engine != null) {
        try {
            return engine.shouldPeriodicallyFlush();
        } catch (final AlreadyClosedException e) {
            // we are already closed, no need to flush or roll
        }
    }
    return false;
}
 

@Test
public void testObtainFileAlreadyClosed() throws IOException {
  replicator.publish(createRevision(1));
  SessionToken res = replicator.checkForUpdate(null);
  assertNotNull(res);
  assertEquals(1, res.sourceFiles.size());
  Entry<String,List<RevisionFile>> entry = res.sourceFiles.entrySet().iterator().next();
  replicator.close();
  expectThrows(AlreadyClosedException.class, () -> {
    replicator.obtainFile(res.id, entry.getKey(), entry.getValue().get(0).fileName);
  });
}
 

@Test
public void testPublishAlreadyClosed() throws IOException {
  replicator.close();
  expectThrows(AlreadyClosedException.class, () -> {
    replicator.publish(createRevision(2));
  });
}
 

@Test
public void testUpdateAlreadyClosed() throws IOException {
  replicator.close();
  expectThrows(AlreadyClosedException.class, () -> {
    replicator.checkForUpdate(null);
  });
}
 

public synchronized void launch(CopyJob job) {
  if (finish == false) {
    queue.offer(job);
    notify();
  } else {
    throw new AlreadyClosedException("closed");
  }
}
 

/**
 * Closes current reader and creates new one with new checkoint and same file channel
 */
TranslogReader closeIntoTrimmedReader(long aboveSeqNo, ChannelFactory channelFactory) throws IOException {
    if (closed.compareAndSet(false, true)) {
        Closeable toCloseOnFailure = channel;
        final TranslogReader newReader;
        try {
            if (aboveSeqNo < checkpoint.trimmedAboveSeqNo
                || aboveSeqNo < checkpoint.maxSeqNo && checkpoint.trimmedAboveSeqNo == SequenceNumbers.UNASSIGNED_SEQ_NO) {
                final Path checkpointFile = path.getParent().resolve(getCommitCheckpointFileName(checkpoint.generation));
                final Checkpoint newCheckpoint = new Checkpoint(checkpoint.offset, checkpoint.numOps,
                    checkpoint.generation, checkpoint.minSeqNo, checkpoint.maxSeqNo,
                    checkpoint.globalCheckpoint, checkpoint.minTranslogGeneration, aboveSeqNo);
                Checkpoint.write(channelFactory, checkpointFile, newCheckpoint, StandardOpenOption.WRITE);

                IOUtils.fsync(checkpointFile, false);
                IOUtils.fsync(checkpointFile.getParent(), true);

                newReader = new TranslogReader(newCheckpoint, channel, path, header);
            } else {
                newReader = new TranslogReader(checkpoint, channel, path, header);
            }
            toCloseOnFailure = null;
            return newReader;
        } finally {
            IOUtils.close(toCloseOnFailure);
        }
    } else {
        throw new AlreadyClosedException(toString() + " is already closed");
    }
}
 

/**
 * Asserts that the document history in Lucene index is consistent with Translog's on every index shard of the cluster.
 * This assertion might be expensive, thus we prefer not to execute on every test but only interesting tests.
 */
public void assertConsistentHistoryBetweenTranslogAndLuceneIndex() throws IOException {
    for (NodeAndClient nodeAndClient : nodes.values()) {
        IndicesService indexServices = getInstance(IndicesService.class, nodeAndClient.name);
        for (IndexService indexService : indexServices) {
            for (IndexShard indexShard : indexService) {
                try {
                    IndexShardTestCase.assertConsistentHistoryBetweenTranslogAndLucene(indexShard);
                } catch (AlreadyClosedException ignored) {
                    // shard is closed
                }
            }
        }
    }
}
 

@Nullable
public Long translogUncommittedSizeInBytes() {
    try {
        var stats = indexShard.translogStats();
        return stats == null ? null : stats.getUncommittedSizeInBytes();
    } catch (AlreadyClosedException e) {
        return 0L;
    }
}