IndexCreatedHandler implements EventHandler { @Override public void handle(Object event) throws Exception { logger.debug("Handler received background indexing event: " + event.toString()); IndexCreatedEvent status = null; try { status = (IndexCreatedEvent) event; Index index = indexRepo.read(status.getIndex().getId()); long offset = 0; long recordsCompleted = 0; QueryResponseWrapper responseWrapper = docRepo.readAll(index.getDatabaseName(), index.getTableName(), CHUNK, offset); boolean hasMore; if (responseWrapper.isEmpty()) { hasMore = false; } else { hasMore = true; } while (hasMore) { for (Document toIndex : responseWrapper) { try { BoundStatement statement = IndexMaintainerHelper.generateDocumentCreateIndexEntryStatement(indexStatusRepo.getSession(), index, toIndex, PrimaryIndexBucketLocatorImpl.getInstance()); if (statement != null) { indexStatusRepo.getSession().execute(statement); } recordsCompleted++; } catch (IndexParseException e) { recordsCompleted++; status.setRecordsCompleted(recordsCompleted); List<String> errors = status.getErrors(); if (errors == null) { errors = new ArrayList<>(); } errors.add(e.toString() + " In document: " + toIndex.toString()); status.setErrors(errors); status.setStatusLastUpdatedAt(new Date()); } } offset = offset + CHUNK; status.setRecordsCompleted(recordsCompleted); status.setStatusLastUpdatedAt(new Date()); indexStatusRepo.update(status); responseWrapper = docRepo.readAll(index.getDatabaseName(), index.getTableName(), CHUNK, offset); if (responseWrapper.isEmpty()) { hasMore = false; } } index.setActive(true); status.setIndex(index); indexRepo.markActive(index); indexStatusRepo.update(status); } catch (Exception e) { String indexName = "Cannot be determined."; if (status != null && status.getIndex() != null) { indexName = status.getIndex().getName(); } String errorMessage = "Could not complete indexing event for index: '" + indexName + "'."; logger.error(errorMessage, e); if (status != null) { status.setFatalError(errorMessage + " Please contact a system administrator to resolve this issue."); status.setStatusLastUpdatedAt(new Date()); indexStatusRepo.update(status); } throw e; } } IndexCreatedHandler(IndexRepository indexRepo, IndexStatusRepository indexStatusRepo, DocumentRepository docRepo); @Override boolean handles(Class<?> eventClass); @Override void handle(Object event);  }
@Test public void testHandle() throws Exception { System.out.println("handle"); Index testIndex = Fixtures.createTestIndexWithBulkDataHit(); f.insertIndex(testIndex); IndexCreatedEvent entity = Fixtures.createTestIndexCreationStatusWithBulkDataHit(); entity.setTotalRecords(34); statusRepo.create(entity); Object event = entity; IndexCreatedHandler instance = new IndexCreatedHandler(indexRepo, statusRepo, docRepo); instance.handle(event); assertTrue(statusRepo.exists(entity.getUuid())); IndexCreatedEvent storedStatus = statusRepo.read(entity.getUuid()); assertNotNull(storedStatus); assertTrue(storedStatus.isDoneIndexing()); assertEquals(storedStatus.getTotalRecords(), storedStatus.getRecordsCompleted()); assertEquals(100, storedStatus.getPercentComplete(), 0); assertEquals(storedStatus.getEta(), 0); assertNotEquals(storedStatus.getDateStarted(), storedStatus.getStatusLastUpdatedAt()); assertTrue(statusRepo.readAllCurrentlyIndexing().isEmpty()); assertNotNull(storedStatus.getIndex()); Index readIndex = indexRepo.read(testIndex.getId()); assertNotNull(readIndex); assertTrue(readIndex.isActive()); }
@Test public void testHandleWithError() throws Exception { System.out.println("handleWithError"); IndexCreatedEvent entity = Fixtures.createTestIndexCreationStatusWithBulkDataHit(); entity.setTotalRecords(34); statusRepo.create(entity); Object event = entity; IndexCreatedHandler instance = new IndexCreatedHandler(indexRepo, statusRepo, docRepo); boolean expectedExceptionThrown = false; try { instance.handle(event); } catch (Exception e) { expectedExceptionThrown = true; } assertTrue("Expected exception not thrown.", expectedExceptionThrown); assertTrue(statusRepo.exists(entity.getUuid())); IndexCreatedEvent storedStatus = statusRepo.read(entity.getUuid()); assertNotNull(storedStatus); assertFalse(storedStatus.isDoneIndexing()); assertNotEquals(storedStatus.getTotalRecords(), storedStatus.getRecordsCompleted()); assertEquals(0, storedStatus.getPercentComplete(), 0); assertEquals(storedStatus.getEta(), -1); assertNotEquals(storedStatus.getDateStarted(), storedStatus.getStatusLastUpdatedAt()); assertFalse(statusRepo.readAllCurrentlyIndexing().isEmpty()); assertEquals("Could not complete indexing event for index: 'myindexbulkdata'. Please contact a system administrator to resolve this issue.", storedStatus.getFatalError()); }
@Test public void testHandleWithData() throws Exception { System.out.println("handleWithData"); Database testDb = Fixtures.createTestWorldBankDatabase(); Table testTable = Fixtures.createTestWorldBankTable(); f.insertDatabase(testDb); f.insertTable(testTable); List<Document> docs = Fixtures.getBulkDocumentsByLine("/world_bank_short.txt", testTable); f.insertDocuments(docs); Index lastname = Fixtures.createTestBankIndexCountryName(); f.insertIndex(lastname); IndexCreatedEvent entity = new IndexCreatedEvent(UUID.randomUUID(), new Date(), new Date(), lastname, docs.size(), 0); statusRepo.create(entity); Object event = entity; IndexCreatedHandler instance = new IndexCreatedHandler(indexRepo, statusRepo, docRepo); instance.handle(event); assertTrue(statusRepo.exists(entity.getUuid())); IndexCreatedEvent storedStatus = statusRepo.read(entity.getUuid()); assertNotNull(storedStatus); assertTrue(storedStatus.isDoneIndexing()); assertEquals(storedStatus.getTotalRecords(), storedStatus.getRecordsCompleted()); assertEquals(100, storedStatus.getPercentComplete(), 0); assertEquals(storedStatus.getEta(), 0); assertNotEquals(storedStatus.getDateStarted(), storedStatus.getStatusLastUpdatedAt()); assertTrue(statusRepo.readAllCurrentlyIndexing().isEmpty()); assertNotNull(storedStatus.getIndex()); Index readIndex = indexRepo.read(lastname.getId()); assertNotNull(readIndex); assertTrue(readIndex.isActive()); }