-
Bug
-
Resolution: Done
-
Major
-
14.0.0.Final
-
None
Soft Index File Store while shutting down does not wait for all the flowables to complete before proceeding. This allows for an operation to be processed in the LogAppender but then it is never sent to the indexer. This causes data to exist without an index to represent it. This means if the server is restarted with a valid index it will not be present, but if the index is rebuilt it will be present.
The following code reproduces it when added to SoftIndexFileStore, which may not be able to function when fixed.
@Test(dataProvider = "booleans") public void testWriteDuringStop(boolean deleteIndexes) throws InterruptedException, TimeoutException, ExecutionException { LogAppender logAppender = TestingUtil.extractField(store.delegate(), "logAppender"); // This will block the write until after it is written to the logAppender but hasn't completed yet, which means // it won't be able to write to the index CheckPoint putCheckPoint = new CheckPoint(); putCheckPoint.triggerForever(Mocks.AFTER_RELEASE); Mocks.blockingFieldMock(putCheckPoint, FlowableProcessor.class, logAppender, LogAppender.class, "completionProcessor", (stub, processor) -> stub.when(processor).onNext(Mockito.any())); Future<Void> putFuture = fork(() -> TestingUtil.join(store.write(1, marshalledEntry(internalCacheEntry("foo-231", "bar-231", -1))))); putCheckPoint.awaitStrict(Mocks.BEFORE_INVOCATION, 10, TimeUnit.SECONDS); TestingUtil.join(store.stop()); putCheckPoint.triggerForever(Mocks.BEFORE_RELEASE); putFuture.get(10, TimeUnit.SECONDS); if (deleteIndexes) { Util.recursiveFileRemove(Paths.get(tmpDirectory, "index")); } startStore(store); assertNotNull("bar-231", TestingUtil.join(store.load(1, "foo-231"))); }
