@Test void runInternal() throws Exception { when(this.configuration.isEventStoreEnabled()).thenReturn(true); when(this.configuration.getEventStore()).thenReturn("myStore"); EventStore eventStore = mock(EventStore.class); when(this.componentManager.getInstance(EventStore.class, "myStore")).thenReturn(eventStore); when(this.eventStream.countEvents()).thenReturn(105L); LegacyEventMigrationRequest request = new LegacyEventMigrationRequest(new Date(42), Arrays.asList("myId", "42")); Query query = mock(Query.class); when(this.queryManager.createQuery("WHERE event.date >= :since", Query.HQL)).thenReturn(query); Event event1 = mock(Event.class); Event event2 = mock(Event.class); Event event3 = mock(Event.class); Event event4 = mock(Event.class); when(this.eventStream.searchEvents(query)).thenReturn(Arrays.asList(event1, event2, event3, event4)); when(event1.getId()).thenReturn("event1"); when(eventStore.getEvent("event1")).thenReturn(Optional.of(event1)); when(event2.getId()).thenReturn("event2"); when(eventStore.getEvent("event2")).thenReturn(Optional.empty()); when(event2.getWiki()).thenReturn(null); when(event3.getId()).thenReturn("event3"); when(eventStore.getEvent("event3")).thenReturn(Optional.empty()); WikiReference event3Wiki = new WikiReference("foo"); when(event3.getWiki()).thenReturn(event3Wiki); when(this.wikiDescriptorManager.exists("foo")).thenReturn(false); when(event4.getId()).thenReturn("event4"); when(eventStore.getEvent("event4")).thenReturn(Optional.empty()); WikiReference event4Wiki = new WikiReference("bar"); when(event4.getWiki()).thenReturn(event4Wiki); when(this.wikiDescriptorManager.exists("bar")).thenReturn(true); CompletableFuture<Event> future = mock(CompletableFuture.class); when(eventStore.saveEvent(event4)).thenReturn(future); this.migrationJob.initialize(request); this.migrationJob.runInternal(); InOrder progressManagerOrder = inOrder(this.progressManager); progressManagerOrder.verify(this.progressManager).pushLevelProgress(eq(2), same(this.migrationJob)); progressManagerOrder.verify(this.progressManager).startStep(same(this.migrationJob)); verify(query).bindValue("since", new Date(42)); verify(query).setOffset(100); verify(eventStore, never()).saveEvent(event1); verify(eventStore).saveEvent(event2); verify(eventStore, never()).saveEvent(event3); verify(eventStore).saveEvent(event4); verify(future).get(); progressManagerOrder.verify(this.progressManager).popLevelProgress(same(this.migrationJob)); }
@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; }