@Test void runInternalNoStoreOrDisabled() throws Exception { when(this.configuration.isEventStoreEnabled()).thenReturn(false); this.migrationJob.runInternal(); assertEquals("New event store system is disabled", logCapture.getMessage(0)); verify(this.componentManager, never()).getInstance(any(), any()); verify(this.eventStream, never()).countEvents(); when(this.configuration.isEventStoreEnabled()).thenReturn(true); when(this.configuration.getEventStore()).thenReturn(""); this.migrationJob.runInternal(); assertEquals("New event store system is disabled", logCapture.getMessage(1)); verify(this.componentManager, never()).getInstance(any(), any()); verify(this.eventStream, never()).countEvents(); when(this.configuration.getEventStore()).thenReturn("myStore"); when(this.componentManager.getInstance(EventStore.class, "myStore")) .thenThrow(new ComponentLookupException("Error when looking for myStore")); this.migrationJob.runInternal(); assertEquals("Failed to get the configured store", logCapture.getMessage(2)); verify(this.componentManager).getInstance(EventStore.class, "myStore"); verify(this.eventStream, never()).countEvents(); }
@Override protected void runInternal() throws Exception { if (!this.configuration.isEventStoreEnabled() || StringUtils.isEmpty(this.configuration.getEventStore())) { this.logger.warn("New event store system is disabled"); return; } try { this.eventStore = this.componentManager.getInstance(EventStore.class, this.configuration.getEventStore()); } catch (ComponentLookupException e) { this.logger.error("Failed to get the configured store", e); return; } long legacyEventCount = this.eventStream.countEvents(); int stepCount = (int) (legacyEventCount / BATCH_SIZE); if (legacyEventCount % BATCH_SIZE != 0) { stepCount++; } this.progressManager.pushLevelProgress(stepCount, this); try { Query query = prepareQuery(); List<Event> events; int offset = 0; do { this.progressManager.startStep(this); events = this.eventStream.searchEvents(query); if (getRequest().isVerbose()) { this.logger.info("Synchronizing legacy events from index {} to {}", offset, offset + events.size()); } if (!events.isEmpty()) { migrate(events); offset += BATCH_SIZE; query.setOffset(offset); } this.progressManager.endStep(this); } while (events.size() == BATCH_SIZE); } finally { this.progressManager.popLevelProgress(this); } }
LegacyEventMigrationJob extends AbstractJob<LegacyEventMigrationRequest, DefaultJobStatus<LegacyEventMigrationRequest>> { @Override protected void runInternal() throws Exception { if (!this.configuration.isEventStoreEnabled() || StringUtils.isEmpty(this.configuration.getEventStore())) { this.logger.warn("New event store system is disabled"); return; } try { this.eventStore = this.componentManager.getInstance(EventStore.class, this.configuration.getEventStore()); } catch (ComponentLookupException e) { this.logger.error("Failed to get the configured store", e); return; } long legacyEventCount = this.eventStream.countEvents(); int stepCount = (int) (legacyEventCount / BATCH_SIZE); if (legacyEventCount % BATCH_SIZE != 0) { stepCount++; } this.progressManager.pushLevelProgress(stepCount, this); try { Query query = prepareQuery(); List<Event> events; int offset = 0; do { this.progressManager.startStep(this); events = this.eventStream.searchEvents(query); if (getRequest().isVerbose()) { this.logger.info("Synchronizing legacy events from index {} to {}", offset, offset + events.size()); } if (!events.isEmpty()) { migrate(events); offset += BATCH_SIZE; query.setOffset(offset); } this.progressManager.endStep(this); } while (events.size() == BATCH_SIZE); } finally { this.progressManager.popLevelProgress(this); } } }
LegacyEventMigrationJob extends AbstractJob<LegacyEventMigrationRequest, DefaultJobStatus<LegacyEventMigrationRequest>> { @Override protected void runInternal() throws Exception { if (!this.configuration.isEventStoreEnabled() || StringUtils.isEmpty(this.configuration.getEventStore())) { this.logger.warn("New event store system is disabled"); return; } try { this.eventStore = this.componentManager.getInstance(EventStore.class, this.configuration.getEventStore()); } catch (ComponentLookupException e) { this.logger.error("Failed to get the configured store", e); return; } long legacyEventCount = this.eventStream.countEvents(); int stepCount = (int) (legacyEventCount / BATCH_SIZE); if (legacyEventCount % BATCH_SIZE != 0) { stepCount++; } this.progressManager.pushLevelProgress(stepCount, this); try { Query query = prepareQuery(); List<Event> events; int offset = 0; do { this.progressManager.startStep(this); events = this.eventStream.searchEvents(query); if (getRequest().isVerbose()) { this.logger.info("Synchronizing legacy events from index {} to {}", offset, offset + events.size()); } if (!events.isEmpty()) { migrate(events); offset += BATCH_SIZE; query.setOffset(offset); } this.progressManager.endStep(this); } while (events.size() == BATCH_SIZE); } finally { this.progressManager.popLevelProgress(this); } }  }
LegacyEventMigrationJob extends AbstractJob<LegacyEventMigrationRequest, DefaultJobStatus<LegacyEventMigrationRequest>> { @Override protected void runInternal() throws Exception { if (!this.configuration.isEventStoreEnabled() || StringUtils.isEmpty(this.configuration.getEventStore())) { this.logger.warn("New event store system is disabled"); return; } try { this.eventStore = this.componentManager.getInstance(EventStore.class, this.configuration.getEventStore()); } catch (ComponentLookupException e) { this.logger.error("Failed to get the configured store", e); return; } long legacyEventCount = this.eventStream.countEvents(); int stepCount = (int) (legacyEventCount / BATCH_SIZE); if (legacyEventCount % BATCH_SIZE != 0) { stepCount++; } this.progressManager.pushLevelProgress(stepCount, this); try { Query query = prepareQuery(); List<Event> events; int offset = 0; do { this.progressManager.startStep(this); events = this.eventStream.searchEvents(query); if (getRequest().isVerbose()) { this.logger.info("Synchronizing legacy events from index {} to {}", offset, offset + events.size()); } if (!events.isEmpty()) { migrate(events); offset += BATCH_SIZE; query.setOffset(offset); } this.progressManager.endStep(this); } while (events.size() == BATCH_SIZE); } finally { this.progressManager.popLevelProgress(this); } }  @Override String getType(); }
LegacyEventMigrationJob extends AbstractJob<LegacyEventMigrationRequest, DefaultJobStatus<LegacyEventMigrationRequest>> { @Override protected void runInternal() throws Exception { if (!this.configuration.isEventStoreEnabled() || StringUtils.isEmpty(this.configuration.getEventStore())) { this.logger.warn("New event store system is disabled"); return; } try { this.eventStore = this.componentManager.getInstance(EventStore.class, this.configuration.getEventStore()); } catch (ComponentLookupException e) { this.logger.error("Failed to get the configured store", e); return; } long legacyEventCount = this.eventStream.countEvents(); int stepCount = (int) (legacyEventCount / BATCH_SIZE); if (legacyEventCount % BATCH_SIZE != 0) { stepCount++; } this.progressManager.pushLevelProgress(stepCount, this); try { Query query = prepareQuery(); List<Event> events; int offset = 0; do { this.progressManager.startStep(this); events = this.eventStream.searchEvents(query); if (getRequest().isVerbose()) { this.logger.info("Synchronizing legacy events from index {} to {}", offset, offset + events.size()); } if (!events.isEmpty()) { migrate(events); offset += BATCH_SIZE; query.setOffset(offset); } this.progressManager.endStep(this); } while (events.size() == BATCH_SIZE); } finally { this.progressManager.popLevelProgress(this); } }  @Override String getType(); static final String JOBTYPE; }