Overview
Projection rebuilding is the process of recreating read models (projections) from event streams or source data. This is crucial in CQRS and Event Sourcing architectures for maintaining consistent read models after schema changes, bug fixes, or system migrations.
Architecture
Projection Rebuilding Strategies
- Full Rebuild: Complete recreation from all events
- Incremental Rebuild: Update from specific point
- Parallel Rebuild: Build new projection while serving old
- Live Migration: Rebuild while system is operational
Dependencies
<dependencies> <!-- Event Store --> <dependency> <groupId>org.axonframework</groupId> <artifactId>axon-core</artifactId> <version>4.8.0</version> </dependency> <!-- Spring Data --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-jpa</artifactId> </dependency> <!-- MongoDB for projections --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-mongodb</artifactId> </dependency> <!-- Redis for cache --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-redis</artifactId> </dependency> <!-- Metrics --> <dependency> <groupId>io.micrometer</groupId> <artifactId>micrometer-core</artifactId> </dependency> <!-- Testing --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope> </dependency> </dependencies>
Core Implementation
1. Projection Interface and Base Class
public interface Projection {
String getName();
long getVersion();
ProjectionStatus getStatus();
Instant getLastRebuildTime();
void rebuild(RebuildContext context);
boolean supportsIncremental();
}
public abstract class AbstractProjection implements Projection {
protected final Logger logger = LoggerFactory.getLogger(getClass());
protected final String name;
protected final long version;
protected volatile ProjectionStatus status = ProjectionStatus.ACTIVE;
protected volatile Instant lastRebuildTime;
protected final AtomicLong processedEvents = new AtomicLong();
protected final AtomicLong rebuildDuration = new AtomicLong();
protected AbstractProjection(String name, long version) {
this.name = name;
this.version = version;
}
@Override
public String getName() { return name; }
@Override
public long getVersion() { return version; }
@Override
public ProjectionStatus getStatus() { return status; }
@Override
public Instant getLastRebuildTime() { return lastRebuildTime; }
@Override
public boolean supportsIncremental() { return false; }
protected void updateStatus(ProjectionStatus newStatus) {
this.status = newStatus;
logger.info("Projection {} status changed to: {}", name, newStatus);
}
protected void recordMetrics(long startTime, long eventCount) {
long duration = System.currentTimeMillis() - startTime;
rebuildDuration.set(duration);
processedEvents.addAndGet(eventCount);
logger.info("Projection {} rebuild completed in {}ms, processed {} events",
name, duration, eventCount);
}
}
2. Projection Rebuilder Service
@Service
public class ProjectionRebuilder {
private final Map<String, Projection> projections;
private final EventStore eventStore;
private final ProjectionRepository projectionRepository;
private final MeterRegistry meterRegistry;
private final ExecutorService rebuildExecutor;
private final Map<String, RebuildContext> activeRebuilds;
public ProjectionRebuilder(EventStore eventStore,
ProjectionRepository projectionRepository,
MeterRegistry meterRegistry,
List<Projection> projectionBeans) {
this.eventStore = eventStore;
this.projectionRepository = projectionRepository;
this.meterRegistry = meterRegistry;
this.projections = projectionBeans.stream()
.collect(Collectors.toMap(Projection::getName, Function.identity()));
this.rebuildExecutor = Executors.newFixedThreadPool(5);
this.activeRebuilds = new ConcurrentHashMap<>();
}
public RebuildResult rebuildProjection(String projectionName, RebuildOptions options) {
Projection projection = projections.get(projectionName);
if (projection == null) {
throw new ProjectionNotFoundException("Projection not found: " + projectionName);
}
if (activeRebuilds.containsKey(projectionName)) {
throw new RebuildInProgressException("Rebuild already in progress for: " + projectionName);
}
return startRebuild(projection, options);
}
public RebuildResult rebuildAll(RebuildOptions options) {
List<RebuildResult> results = projections.values().stream()
.map(projection -> startRebuild(projection, options))
.collect(Collectors.toList());
return RebuildResult.aggregate(results);
}
private RebuildResult startRebuild(Projection projection, RebuildOptions options) {
RebuildContext context = createRebuildContext(projection, options);
activeRebuilds.put(projection.getName(), context);
CompletableFuture<RebuildResult> future = CompletableFuture.supplyAsync(() -> {
try {
return executeRebuild(projection, context);
} finally {
activeRebuilds.remove(projection.getName());
}
}, rebuildExecutor);
context.setFuture(future);
return RebuildResult.started(projection.getName(), context.getRebuildId());
}
private RebuildResult executeRebuild(Projection projection, RebuildContext context) {
long startTime = System.currentTimeMillis();
projection.updateStatus(ProjectionStatus.REBUILDING);
try {
// Create backup if needed
if (context.getOptions().isBackupEnabled()) {
createBackup(projection);
}
// Execute rebuild
projection.rebuild(context);
// Update metadata
updateProjectionMetadata(projection, context);
// Record success
recordRebuildMetrics(projection, startTime, context.getProcessedEvents(), true);
return RebuildResult.completed(
projection.getName(),
context.getRebuildId(),
context.getProcessedEvents(),
System.currentTimeMillis() - startTime
);
} catch (Exception e) {
// Handle failure
handleRebuildFailure(projection, context, e);
recordRebuildMetrics(projection, startTime, context.getProcessedEvents(), false);
return RebuildResult.failed(
projection.getName(),
context.getRebuildId(),
e.getMessage(),
System.currentTimeMillis() - startTime
);
} finally {
projection.updateStatus(ProjectionStatus.ACTIVE);
}
}
private RebuildContext createRebuildContext(Projection projection, RebuildOptions options) {
return RebuildContext.builder()
.rebuildId(UUID.randomUUID().toString())
.projectionName(projection.getName())
.options(options)
.startTime(Instant.now())
.build();
}
private void createBackup(Projection projection) {
try {
projectionRepository.createBackup(projection.getName());
logger.info("Created backup for projection: {}", projection.getName());
} catch (Exception e) {
logger.warn("Failed to create backup for projection: {}", projection.getName(), e);
}
}
private void updateProjectionMetadata(Projection projection, RebuildContext context) {
ProjectionMetadata metadata = new ProjectionMetadata(
projection.getName(),
projection.getVersion(),
Instant.now(),
context.getProcessedEvents(),
context.getOptions().getRebuildType()
);
projectionRepository.saveMetadata(metadata);
}
private void recordRebuildMetrics(Projection projection, long startTime,
long eventCount, boolean success) {
long duration = System.currentTimeMillis() - startTime;
meterRegistry.timer("projection.rebuild.duration",
"projection", projection.getName())
.record(Duration.ofMillis(duration));
meterRegistry.counter("projection.rebuild.events",
"projection", projection.getName())
.increment(eventCount);
meterRegistry.counter("projection.rebuild.outcome",
"projection", projection.getName(),
"success", String.valueOf(success))
.increment();
}
private void handleRebuildFailure(Projection projection, RebuildContext context, Exception e) {
logger.error("Rebuild failed for projection: {}", projection.getName(), e);
// Restore from backup if available
if (context.getOptions().isBackupEnabled()) {
try {
projectionRepository.restoreBackup(projection.getName());
logger.info("Restored projection from backup: {}", projection.getName());
} catch (Exception restoreException) {
logger.error("Failed to restore projection from backup: {}",
projection.getName(), restoreException);
}
}
}
public RebuildStatus getRebuildStatus(String projectionName) {
RebuildContext context = activeRebuilds.get(projectionName);
if (context != null) {
return RebuildStatus.inProgress(
context.getRebuildId(),
context.getProcessedEvents(),
context.getStartTime()
);
}
ProjectionMetadata metadata = projectionRepository.getMetadata(projectionName);
return RebuildStatus.completed(metadata);
}
public void cancelRebuild(String projectionName) {
RebuildContext context = activeRebuilds.get(projectionName);
if (context != null && context.getFuture() != null) {
context.getFuture().cancel(true);
activeRebuilds.remove(projectionName);
}
}
}
3. Event-Based Projection Implementation
@Component
public class OrderSummaryProjection extends AbstractProjection {
private final OrderSummaryRepository repository;
private final EventStore eventStore;
public OrderSummaryProjection(OrderSummaryRepository repository,
EventStore eventStore) {
super("order-summary", 2L);
this.repository = repository;
this.eventStore = eventStore;
}
@Override
public void rebuild(RebuildContext context) {
long startTime = System.currentTimeMillis();
long eventCount = 0;
try {
// Clear existing data
if (context.getOptions().getRebuildType() == RebuildType.FULL) {
repository.deleteAll();
}
// Rebuild from events
eventCount = processEvents(context);
// Update any derived data
updateDerivedData();
} catch (Exception e) {
throw new ProjectionRebuildException("Failed to rebuild order summary projection", e);
} finally {
recordMetrics(startTime, eventCount);
}
}
@Override
public boolean supportsIncremental() {
return true;
}
private long processEvents(RebuildContext context) {
AtomicLong eventCount = new AtomicLong();
eventStore.readAllEvents(context.getOptions().getFromSequence())
.takeWhile(event -> shouldContinueProcessing(context))
.forEach(event -> {
try {
handleEvent(event, context);
eventCount.incrementAndGet();
context.incrementProcessedEvents();
// Progress reporting
if (eventCount.get() % 1000 == 0) {
logger.info("Processed {} events for projection {}",
eventCount.get(), getName());
}
} catch (Exception e) {
if (context.getOptions().isFailOnError()) {
throw new ProjectionRebuildException(
"Error processing event: " + event.getSequence(), e);
} else {
logger.warn("Skipped event {} due to error: {}",
event.getSequence(), e.getMessage());
}
}
});
return eventCount.get();
}
private void handleEvent(DomainEvent event, RebuildContext context) {
switch (event.getType()) {
case "OrderCreated":
handleOrderCreated((OrderCreated) event);
break;
case "OrderCancelled":
handleOrderCancelled((OrderCancelled) event);
break;
case "OrderShipped":
handleOrderShipped((OrderShipped) event);
break;
case "OrderDelivered":
handleOrderDelivered((OrderDelivered) event);
break;
default:
// Ignore unknown event types
break;
}
}
private void handleOrderCreated(OrderCreated event) {
OrderSummary summary = new OrderSummary();
summary.setOrderId(event.getOrderId());
summary.setCustomerId(event.getCustomerId());
summary.setAmount(event.getAmount());
summary.setStatus("CREATED");
summary.setCreatedAt(event.getTimestamp());
summary.setUpdatedAt(event.getTimestamp());
repository.save(summary);
}
private void handleOrderCancelled(OrderCancelled event) {
repository.findById(event.getOrderId()).ifPresent(summary -> {
summary.setStatus("CANCELLED");
summary.setUpdatedAt(event.getTimestamp());
repository.save(summary);
});
}
private void handleOrderShipped(OrderShipped event) {
repository.findById(event.getOrderId()).ifPresent(summary -> {
summary.setStatus("SHIPPED");
summary.setShippingDate(event.getTimestamp());
summary.setUpdatedAt(event.getTimestamp());
repository.save(summary);
});
}
private void handleOrderDelivered(OrderDelivered event) {
repository.findById(event.getOrderId()).ifPresent(summary -> {
summary.setStatus("DELIVERED");
summary.setDeliveryDate(event.getTimestamp());
summary.setUpdatedAt(event.getTimestamp());
repository.save(summary);
});
}
private void updateDerivedData() {
// Update any aggregated data
updateCustomerStatistics();
updateProductStatistics();
}
private void updateCustomerStatistics() {
// Implementation for updating customer stats
}
private void updateProductStatistics() {
// Implementation for updating product stats
}
private boolean shouldContinueProcessing(RebuildContext context) {
return context.getOptions().getMaxEvents() <= 0 ||
context.getProcessedEvents() < context.getOptions().getMaxEvents();
}
}
Advanced Rebuilding Strategies
1. Parallel Rebuild with Switchover
@Service
public class ParallelProjectionRebuilder {
private final ProjectionRebuilder rebuilder;
private final ProjectionRouter projectionRouter;
private final Map<String, String> activeVersions;
public ParallelProjectionRebuilder(ProjectionRebuilder rebuilder,
ProjectionRouter projectionRouter) {
this.rebuilder = rebuilder;
this.projectionRouter = projectionRouter;
this.activeVersions = new ConcurrentHashMap<>();
}
public ParallelRebuildResult rebuildInParallel(String projectionName,
RebuildOptions options) {
String newVersion = generateVersionId();
String tempProjectionName = projectionName + "-" + newVersion;
// Create temporary projection instance
Projection tempProjection = createTemporaryProjection(projectionName, tempProjectionName);
// Register temporary projection for routing
projectionRouter.registerVersion(projectionName, newVersion, tempProjectionName);
// Start rebuild to temporary projection
RebuildResult result = rebuilder.rebuildProjection(tempProjectionName, options);
return new ParallelRebuildResult(
projectionName,
newVersion,
tempProjectionName,
result
);
}
public void switchToVersion(String projectionName, String version) {
projectionRouter.setActiveVersion(projectionName, version);
activeVersions.put(projectionName, version);
// Clean up old versions
cleanupOldVersions(projectionName, version);
}
public void cancelParallelRebuild(String projectionName, String version) {
projectionRouter.unregisterVersion(projectionName, version);
rebuilder.cancelRebuild(projectionName + "-" + version);
}
private Projection createTemporaryProjection(String originalName, String tempName) {
// Clone or create temporary projection with different storage
// This would typically use a different database/schema
return new TemporaryProjection(originalName, tempName);
}
private String generateVersionId() {
return Instant.now().toString().replaceAll("[:.-]", "").toLowerCase();
}
private void cleanupOldVersions(String projectionName, String currentVersion) {
projectionRouter.getAllVersions(projectionName).stream()
.filter(version -> !version.equals(currentVersion))
.forEach(version -> {
projectionRouter.unregisterVersion(projectionName, version);
// Schedule deletion of old projection data
scheduleProjectionCleanup(projectionName + "-" + version);
});
}
private void scheduleProjectionCleanup(String projectionName) {
// Schedule cleanup of old projection data
}
}
2. Incremental Rebuild Service
@Service
public class IncrementalRebuildService {
private final EventStore eventStore;
private final ProjectionMetadataRepository metadataRepository;
private final Map<String, Projection> projections;
public IncrementalRebuildService(EventStore eventStore,
ProjectionMetadataRepository metadataRepository,
List<Projection> projectionBeans) {
this.eventStore = eventStore;
this.metadataRepository = metadataRepository;
this.projections = projectionBeans.stream()
.collect(Collectors.toMap(Projection::getName, Function.identity()));
}
@Scheduled(fixedDelay = 30000) // Every 30 seconds
public void performIncrementalRebuilds() {
projections.values().stream()
.filter(Projection::supportsIncremental)
.forEach(this::rebuildIncrementally);
}
private void rebuildIncrementally(Projection projection) {
ProjectionMetadata metadata = metadataRepository.findByName(projection.getName());
long lastProcessedSequence = metadata.getLastProcessedSequence();
try {
long newSequence = processNewEvents(projection, lastProcessedSequence);
if (newSequence > lastProcessedSequence) {
metadata.setLastProcessedSequence(newSequence);
metadata.setLastUpdated(Instant.now());
metadataRepository.save(metadata);
logger.debug("Incremental rebuild completed for {} up to sequence {}",
projection.getName(), newSequence);
}
} catch (Exception e) {
logger.error("Incremental rebuild failed for {}", projection.getName(), e);
}
}
private long processNewEvents(Projection projection, long lastProcessedSequence) {
AtomicLong latestSequence = new AtomicLong(lastProcessedSequence);
eventStore.readEventsAfter(lastProcessedSequence)
.forEach(event -> {
try {
handleEventForProjection(projection, event);
latestSequence.set(event.getSequence());
} catch (Exception e) {
logger.warn("Failed to process event {} for projection {}",
event.getSequence(), projection.getName(), e);
}
});
return latestSequence.get();
}
private void handleEventForProjection(Projection projection, DomainEvent event) {
// This would need reflection or a registry of event handlers
if (projection instanceof EventHandler) {
((EventHandler) projection).handleEvent(event);
}
}
public void triggerIncrementalRebuild(String projectionName) {
Projection projection = projections.get(projectionName);
if (projection != null && projection.supportsIncremental()) {
rebuildIncrementally(projection);
}
}
}
Monitoring and Management
1. Projection Rebuild Controller
@RestController
@RequestMapping("/api/projections")
public class ProjectionRebuildController {
private final ProjectionRebuilder rebuilder;
private final ParallelProjectionRebuilder parallelRebuilder;
private final IncrementalRebuildService incrementalService;
public ProjectionRebuildController(ProjectionRebuilder rebuilder,
ParallelProjectionRebuilder parallelRebuilder,
IncrementalRebuildService incrementalService) {
this.rebuilder = rebuilder;
this.parallelRebuilder = parallelRebuilder;
this.incrementalService = incrementalService;
}
@PostMapping("/{projectionName}/rebuild")
public ResponseEntity<RebuildResult> rebuildProjection(
@PathVariable String projectionName,
@RequestBody RebuildRequest request) {
RebuildOptions options = RebuildOptions.builder()
.rebuildType(request.getRebuildType())
.fromSequence(request.getFromSequence())
.maxEvents(request.getMaxEvents())
.backupEnabled(request.isBackupEnabled())
.failOnError(request.isFailOnError())
.build();
RebuildResult result = rebuilder.rebuildProjection(projectionName, options);
return ResponseEntity.accepted().body(result);
}
@PostMapping("/{projectionName}/rebuild-parallel")
public ResponseEntity<ParallelRebuildResult> rebuildParallel(
@PathVariable String projectionName,
@RequestBody RebuildRequest request) {
RebuildOptions options = RebuildOptions.builder()
.rebuildType(request.getRebuildType())
.fromSequence(request.getFromSequence())
.maxEvents(request.getMaxEvents())
.backupEnabled(request.isBackupEnabled())
.failOnError(request.isFailOnError())
.build();
ParallelRebuildResult result = parallelRebuilder.rebuildInParallel(projectionName, options);
return ResponseEntity.accepted().body(result);
}
@PostMapping("/{projectionName}/switch-version/{version}")
public ResponseEntity<Void> switchVersion(
@PathVariable String projectionName,
@PathVariable String version) {
parallelRebuilder.switchToVersion(projectionName, version);
return ResponseEntity.ok().build();
}
@GetMapping("/{projectionName}/status")
public ResponseEntity<RebuildStatus> getRebuildStatus(
@PathVariable String projectionName) {
RebuildStatus status = rebuilder.getRebuildStatus(projectionName);
return ResponseEntity.ok(status);
}
@PostMapping("/{projectionName}/cancel")
public ResponseEntity<Void> cancelRebuild(@PathVariable String projectionName) {
rebuilder.cancelRebuild(projectionName);
return ResponseEntity.ok().build();
}
@PostMapping("/{projectionName}/incremental")
public ResponseEntity<Void> triggerIncremental(@PathVariable String projectionName) {
incrementalService.triggerIncrementalRebuild(projectionName);
return ResponseEntity.accepted().build();
}
@PostMapping("/rebuild-all")
public ResponseEntity<RebuildResult> rebuildAll(@RequestBody RebuildRequest request) {
RebuildOptions options = RebuildOptions.builder()
.rebuildType(request.getRebuildType())
.fromSequence(request.getFromSequence())
.maxEvents(request.getMaxEvents())
.backupEnabled(request.isBackupEnabled())
.failOnError(request.isFailOnError())
.build();
RebuildResult result = rebuilder.rebuildAll(options);
return ResponseEntity.accepted().body(result);
}
}
2. Projection Health Check
@Component
public class ProjectionHealthIndicator implements HealthIndicator {
private final ProjectionRebuilder rebuilder;
private final ProjectionMetadataRepository metadataRepository;
public ProjectionHealthIndicator(ProjectionRebuilder rebuilder,
ProjectionMetadataRepository metadataRepository) {
this.rebuilder = rebuilder;
this.metadataRepository = metadataRepository;
}
@Override
public Health health() {
Map<String, Object> details = new HashMap<>();
boolean healthy = true;
try {
List<ProjectionMetadata> allMetadata = metadataRepository.findAll();
for (ProjectionMetadata metadata : allMetadata) {
RebuildStatus status = rebuilder.getRebuildStatus(metadata.getName());
details.put(metadata.getName(), Map.of(
"version", metadata.getVersion(),
"lastRebuild", metadata.getLastRebuildTime(),
"status", status.getStatus(),
"eventsProcessed", metadata.getEventsProcessed()
));
if (status.getStatus() == RebuildStatus.Status.FAILED) {
healthy = false;
}
// Check if projection is stale (no updates in 24 hours)
if (isProjectionStale(metadata)) {
details.put(metadata.getName() + ".stale", true);
healthy = false;
}
}
} catch (Exception e) {
return Health.down(e).build();
}
return healthy ? Health.up().withDetails(details).build() :
Health.down().withDetails(details).build();
}
private boolean isProjectionStale(ProjectionMetadata metadata) {
return metadata.getLastRebuildTime()
.isBefore(Instant.now().minus(24, ChronoUnit.HOURS));
}
}
Testing
1. Projection Rebuild Test
@SpringBootTest
class ProjectionRebuildTest {
@Autowired
private ProjectionRebuilder rebuilder;
@Autowired
private OrderSummaryRepository repository;
@Autowired
private EventStore eventStore;
@Test
void testFullRebuild() {
// Given
RebuildOptions options = RebuildOptions.builder()
.rebuildType(RebuildType.FULL)
.backupEnabled(true)
.build();
// When
RebuildResult result = rebuilder.rebuildProjection("order-summary", options);
// Then
assertEquals(RebuildResult.Status.COMPLETED, result.getStatus());
assertTrue(result.getProcessedEvents() > 0);
// Verify projection data
List<OrderSummary> summaries = repository.findAll();
assertFalse(summaries.isEmpty());
}
@Test
void testRebuildWithEventFiltering() {
// Given
long specificSequence = 1000L;
RebuildOptions options = RebuildOptions.builder()
.rebuildType(RebuildType.FULL)
.fromSequence(specificSequence)
.maxEvents(500)
.build();
// When
RebuildResult result = rebuilder.rebuildProjection("order-summary", options);
// Then
assertEquals(RebuildResult.Status.COMPLETED, result.getStatus());
assertTrue(result.getProcessedEvents() <= 500);
}
@Test
void testRebuildFailure() {
// Given - corrupt event data
RebuildOptions options = RebuildOptions.builder()
.rebuildType(RebuildType.FULL)
.failOnError(true)
.build();
// When/Then
assertThrows(ProjectionRebuildException.class, () -> {
rebuilder.rebuildProjection("corrupted-projection", options);
});
}
}
Configuration
application.yml
projection: rebuild: default-options: backup-enabled: true fail-on-error: false max-events: 0 # 0 means no limit parallel: enabled: true max-concurrent: 3 incremental: enabled: true interval: 30000 # 30 seconds monitoring: health-check-enabled: true metrics-enabled: true events: store: type: jpa batch-size: 1000 read-timeout: 30000 spring: data: mongodb: database: projections jpa: show-sql: false
Best Practices
- Always Backup: Enable backups before major rebuilds
- Monitor Progress: Track rebuild progress and metrics
- Handle Failures Gracefully: Implement proper error handling and recovery
- Test Thoroughly: Test rebuild scenarios with various data sizes
- Use Parallel Rebuilds: For large datasets, use parallel rebuilds to minimize downtime
- Implement Incremental Updates: For frequently changing data
- Version Projections: Maintain version history for rollback capability
- Monitor Performance: Track rebuild duration and resource usage
This implementation provides a comprehensive foundation for projection rebuilding in Java, supporting various rebuild strategies, monitoring, and management capabilities suitable for production systems.