intermediate~2h

Persistence Layer

This is where consumed events stop being ephemeral. Postgres, Flyway, and the NEW-vs-UPDATE logic that makes the consumer more than a dumb logger.

Learning objectives

Beginner: Persist NEW events as straightforward inserts, accepting that a duplicate redelivery would currently fail on a primary-key collision (a known gap to close next).
Intermediate: Make save() upsert-safe (as shown above) so redelivery of the same event is a no-op rather than an error.
Advanced: Combine idempotent persistence here with the producer's idempotent send (Module 13) and consumer transactions (Module 16) for end-to-end exactly-once effect despite at-least-once delivery at every hop.

spring:
  datasource:
    url: jdbc:postgresql://localhost:5432/library_events
    username: postgres
    password: postgres
  jpa:
    hibernate:
      ddl-auto: validate   # schema owned by Flyway, not Hibernate — see §10.2

💻 Code example

spring:
  datasource:
    url: jdbc:postgresql://localhost:5432/library_events
    username: postgres
    password: postgres
  jpa:
    hibernate:
      ddl-auto: validate   # schema owned by Flyway, not Hibernate — see §10.2

◆ The problem

Letting Hibernate auto-generate/alter your schema (ddl-auto: update) is convenient in a demo and dangerous in a team codebase — nobody can tell from source control what actually happened to the schema, and Hibernate's auto-DDL is not always safe against production data.

Flyway replaces that with explicit, versioned, checked-in SQL files that run in order and are tracked in a flyway_schema_history table — the schema's history becomes as reviewable as any other code change.

CREATE TABLE book (
    book_id     INTEGER PRIMARY KEY,
    book_name   VARCHAR(255) NOT NULL,
    book_author VARCHAR(255) NOT NULL
);

CREATE TABLE library_event (
    library_event_id   SERIAL PRIMARY KEY,
    library_event_type VARCHAR(20) NOT NULL,
    book_id             INTEGER REFERENCES book(book_id)
);

▲ Pitfall

Never edit a migration file that has already run in any shared environment — Flyway checksums each applied migration and will refuse to run (or flag drift) if a previously-applied file's content changes. Always add a new V2__...sql instead, even to fix a mistake in V1. This is exactly the rule encoded in the AGENTS.md example from Module 05.

💻 Code example

CREATE TABLE book (
    book_id     INTEGER PRIMARY KEY,
    book_name   VARCHAR(255) NOT NULL,
    book_author VARCHAR(255) NOT NULL
);

CREATE TABLE library_event (
    library_event_id   SERIAL PRIMARY KEY,
    library_event_type VARCHAR(20) NOT NULL,
    book_id             INTEGER REFERENCES book(book_id)
);

@Entity
public class LibraryEventEntity {
    @Id @GeneratedValue private Integer libraryEventId;
    @Enumerated(EnumType.STRING) private LibraryEventType libraryEventType;
    @OneToOne(cascade = { CascadeType.ALL }) private BookEntity book;
}

public interface LibraryEventRepository extends JpaRepository<LibraryEventEntity, Integer> {}

💻 Code example

@Entity
public class LibraryEventEntity {
    @Id @GeneratedValue private Integer libraryEventId;
    @Enumerated(EnumType.STRING) private LibraryEventType libraryEventType;
    @OneToOne(cascade = { CascadeType.ALL }) private BookEntity book;
}

public interface LibraryEventRepository extends JpaRepository<LibraryEventEntity, Integer> {}

public void processLibraryEvent(ConsumerRecord<String, String> record) throws JsonProcessingException {
    LibraryEvent event = objectMapper.readValue(record.value(), LibraryEvent.class);
    switch (event.libraryEventType()) {
        case NEW -> save(event);
        case UPDATE -> {
            if (event.libraryEventId() == null)
                throw new IllegalArgumentException("UPDATE event missing libraryEventId");
            libraryEventRepository.findById(event.libraryEventId())
                .orElseThrow(() -> new IllegalArgumentException("no such libraryEventId"));
            save(event);
        }
    }
}

◆ Under the hood — why this matters for retry safety

Because Kafka consumption is at-least-once (§09.5), this method can run twice for the same record. Notice the save() call for both branches uses the event's own libraryEventId — a JPA save with an existing ID performs an update, not a duplicate insert, so reprocessing the same NEW event twice is safe (idempotent) as long as the ID is deterministic and preserved across redelivery. This is a design decision worth making deliberately, not an accident — see the beginner/intermediate/advanced ladder below.

💻 Code example

public void processLibraryEvent(ConsumerRecord<String, String> record) throws JsonProcessingException {
    LibraryEvent event = objectMapper.readValue(record.value(), LibraryEvent.class);
    switch (event.libraryEventType()) {
        case NEW -> save(event);
        case UPDATE -> {
            if (event.libraryEventId() == null)
                throw new IllegalArgumentException("UPDATE event missing libraryEventId");
            libraryEventRepository.findById(event.libraryEventId())
                .orElseThrow(() -> new IllegalArgumentException("no such libraryEventId"));
            save(event);
        }
    }
}

With both services and Postgres running locally, POST a LibraryEvent to the producer, then query the consumer's database directly — the full path from §04.1's diagram, actually exercised: REST → KafkaTemplate → topic → @KafkaListener → service → JPA → Postgres.

✓ Quick recap

Why use Flyway instead of Hibernate's ddl-auto? Explicit, versioned, reviewable schema history instead of implicit auto-generated DDL. What must you never do to an already-applied Flyway migration? Edit it — Flyway checksums applied migrations and flags/rejects changes to them; add a new migration instead. How does reusing the event's own ID on save() help with Kafka's at-least-once delivery? It makes reprocessing the same record an update-in-place rather than a duplicate row.

Want a visual for this concept?

Generate a diagram tailored to “Persistence Layer” — the AI picks whichever visual (flowchart, comparison, sequence, etc.) best fits.

Practice quiz

Next Step

Continue to Testing the Consumer with Embedded Kafka →← Back to all Kafka & Microservices chapters