Spring Integration JPA

Spring Integration JPA provides comprehensive Java Persistence API (JPA) integration components that enable message-driven database operations within Spring Integration flows. It supports polling entities from databases, persisting and merging entities, executing update queries, and retrieving data using JPQL, native SQL, or named queries.

Key Information for Agents

Required Dependencies:

• spring-integration-jpa (this package)
• spring-integration-core is required
• JPA implementation (Hibernate, EclipseLink, etc.) must be on classpath
• EntityManagerFactory or EntityManager bean must be configured
• spring-orm for JPA support

Default Behaviors:

• Inbound adapters: expectSingleResult=false (returns List), maxResults=unlimited, deleteAfterPoll=false
• Outbound adapters: persistMode=MERGE, flush=false, flushSize=0 (no batch flushing)
• Gateways: expectReply=true (request-reply pattern)
• Query execution: Uses JpaExecutor which manages EntityManager lifecycle
• Transaction participation: Components participate in Spring transactions when configured

Threading Model:

• Inbound adapters: Execute on poller thread (configurable via poller)
• Outbound components: Execute on message processing thread
• EntityManagerFactory: Thread-safe (recommended for concurrent access)
• EntityManager: Not thread-safe (use only in single-threaded contexts)
• Batch operations: Use flushSize for efficient bulk processing

Lifecycle:

• Components initialized when Spring context starts
• JpaExecutor.afterPropertiesSet() must be called after configuration
• Inbound adapters start polling when context is active
• Components stop gracefully on context shutdown

Exceptions:

• JpaOperationFailedException - JPA operation failures (contains offending query)
• PersistenceException - General JPA persistence errors
• EntityExistsException - Entity already exists (when using PERSIST)
• IllegalArgumentException - Invalid configuration or parameters
• NoResultException - Query returned no results (when expectSingleResult=true)

Edge Cases:

• Native queries require entityClass for result mapping
• Named queries must be defined on entity classes with @NamedQuery
• deleteAfterPoll with deleteInBatch=true is more efficient for large batches
• flushSize > 0 enables batch flushing (flushes every N operations)
• clearOnFlush=true prevents memory issues with large entity sets
• ID-based retrieval (via idExpression) uses EntityManager.find() for optimal performance
• When expectSingleResult=true and multiple results found, throws exception
• When expectSingleResult=false and no results, returns empty list (not null)
• Parameter sources can mix static values, expressions, and bean properties

Package Information

• Package Name: spring-integration-jpa
• Package Group: org.springframework.integration
• Package Type: maven
• Language: Java
• Version: 7.0.0
• Installation: Add dependency to Maven/Gradle:

Maven:

<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-jpa</artifactId>
    <version>7.0.0</version>
</dependency>

Gradle:

implementation 'org.springframework.integration:spring-integration-jpa:7.0.0'

Core Imports

Java DSL (recommended):

import org.springframework.integration.jpa.dsl.Jpa;
import org.springframework.integration.jpa.support.PersistMode;
import org.springframework.integration.jpa.support.OutboundGatewayType;

Programmatic configuration:

import org.springframework.integration.jpa.core.JpaExecutor;
import org.springframework.integration.jpa.inbound.JpaPollingChannelAdapter;
import org.springframework.integration.jpa.outbound.JpaOutboundGateway;

XML configuration namespace:

xmlns:int-jpa="http://www.springframework.org/schema/integration/jpa"

Basic Usage

Polling Entities from Database

@Configuration
@EnableIntegration
public class JpaInboundConfig {

    @Bean
    public IntegrationFlow jpaInboundFlow(EntityManagerFactory entityManagerFactory) {
        return IntegrationFlow
            .from(Jpa.inboundAdapter(entityManagerFactory)
                    .entityClass(Student.class)
                    .maxResults(10)
                    .deleteAfterPoll(true),
                e -> e.poller(Pollers.fixedDelay(5000)))
            .channel("studentChannel")
            .get();
    }
}

Persisting Entities to Database

@Bean
public IntegrationFlow jpaOutboundFlow(EntityManagerFactory entityManagerFactory) {
    return IntegrationFlow
        .from("inputChannel")
        .handle(Jpa.outboundAdapter(entityManagerFactory)
                .entityClass(Student.class)
                .persistMode(PersistMode.PERSIST))
        .get();
}

Retrieving Data with Query

@Bean
public IntegrationFlow jpaRetrievingGatewayFlow(EntityManagerFactory entityManagerFactory) {
    return IntegrationFlow
        .from("queryChannel")
        .handle(Jpa.retrievingGateway(entityManagerFactory)
                .jpaQuery("SELECT s FROM Student s WHERE s.lastName = :lastName")
                .parameterExpression("lastName", "payload['lastName']")
                .expectSingleResult(false))
        .channel("resultChannel")
        .get();
}

Architecture

Spring Integration JPA is organized around four primary integration components:

1. Inbound Channel Adapter - Polls database using JPA queries and produces messages
2. Outbound Channel Adapter - Persists/merges/deletes entities (fire-and-forget)
3. Retrieving Outbound Gateway - Queries database and returns results to reply channel
4. Updating Outbound Gateway - Persists/merges entities and returns them to reply channel

All components are powered by JpaExecutor, the central execution engine that handles both retrieval and persistence operations. Configuration can be done via Java DSL (recommended), programmatic configuration, or XML namespace.

Capabilities

Inbound Channel Adapter

Polls the database using JPA queries and produces messages containing retrieved entities.

// Java DSL
static JpaInboundChannelAdapterSpec Jpa.inboundAdapter(EntityManagerFactory entityManagerFactory)
static JpaInboundChannelAdapterSpec Jpa.inboundAdapter(EntityManager entityManager)
static JpaInboundChannelAdapterSpec Jpa.inboundAdapter(JpaOperations jpaOperations)

Key capabilities:

• Poll entities by entity class, JPQL query, native query, or named query
• Delete entities after polling (individually or in batch)
• Limit result size and expect single results
• Flush EntityManager after deletion

Inbound Channel Adapter

Outbound Channel Adapter

Persists, merges, or deletes entities from incoming messages without returning results.

// Java DSL
static JpaUpdatingOutboundEndpointSpec Jpa.outboundAdapter(EntityManagerFactory entityManagerFactory)
static JpaUpdatingOutboundEndpointSpec Jpa.outboundAdapter(EntityManager entityManager)
static JpaUpdatingOutboundEndpointSpec Jpa.outboundAdapter(JpaOperations jpaOperations)

Key capabilities:

• Persist, merge, or delete entities
• Execute parameterized update queries
• Flush EntityManager after operations
• Use message payload or expressions as parameters

Outbound Channel Adapter

Retrieving Outbound Gateway

Queries the database and sends results to a reply channel for continued processing.

// Java DSL
static JpaRetrievingOutboundGatewaySpec Jpa.retrievingGateway(EntityManagerFactory entityManagerFactory)
static JpaRetrievingOutboundGatewaySpec Jpa.retrievingGateway(EntityManager entityManager)
static JpaRetrievingOutboundGatewaySpec Jpa.retrievingGateway(JpaOperations jpaOperations)

Key capabilities:

• Execute SELECT queries (JPQL, native SQL, or named queries)
• Retrieve entities by ID using expressions
• Support pagination (first result, max results)
• Delete retrieved entities after poll
• Return single entity or list of entities

Retrieving Outbound Gateway

Updating Outbound Gateway

Persists or merges entities and returns the persisted/merged entities to a reply channel.

// Java DSL
static JpaUpdatingOutboundEndpointSpec Jpa.updatingGateway(EntityManagerFactory entityManagerFactory)
static JpaUpdatingOutboundEndpointSpec Jpa.updatingGateway(EntityManager entityManager)
static JpaUpdatingOutboundEndpointSpec Jpa.updatingGateway(JpaOperations jpaOperations)

Key capabilities:

• Persist or merge entities with result returned
• Execute update queries and return affected row count
• Support for batch flushing
• Clear persistence context after flush

Updating Outbound Gateway

JPA Executor

The central execution engine that performs JPA operations for all components.

class JpaExecutor {
    JpaExecutor(EntityManagerFactory entityManagerFactory)
    JpaExecutor(EntityManager entityManager)
    JpaExecutor(JpaOperations jpaOperations)
}

Key capabilities:

• Configure query type (entity class, JPQL, native, named query)
• Set parameter sources and parameter expressions
• Execute persistence operations (persist, merge, delete)
• Execute retrieval operations (find, query)
• Support transactional operations

JPA Executor

Parameter Sources

Provides flexible parameter binding for JPA queries using static values, bean properties, or SpEL expressions.

interface ParameterSourceFactory {
    ParameterSource createParameterSource(Object input)
}

class BeanPropertyParameterSourceFactory implements ParameterSourceFactory
class ExpressionEvaluatingParameterSourceFactory implements ParameterSourceFactory

Key capabilities:

• Extract parameters from bean properties
• Evaluate SpEL expressions for parameter values
• Support static parameter values
• Mix static and dynamic parameters

Parameter Sources

JPA Operations

Core interface defining all JPA operations, with default implementation.

interface JpaOperations {
    void persist(Object entity)
    void persist(Object entity, int flushSize, boolean clearOnFlush)
    Object merge(Object entity)
    Object merge(Object entity, int flushSize, boolean clearOnFlush)
    void delete(Object entity)
    void deleteInBatch(Iterable<?> entities)
    <T> T find(Class<T> entityClass, Object primaryKey)
    List<?> getResultListForQuery(String query, ParameterSource source, int firstResult, int maxNumberOfResults)
    // ... additional query and update methods
    void flush()
}

class DefaultJpaOperations implements JpaOperations

JPA Operations

XML Configuration

Spring XML namespace support for declaring JPA components.

<int-jpa:inbound-channel-adapter
    entity-manager-factory="entityManagerFactory"
    entity-class="com.example.Student"
    channel="outputChannel">
    <int:poller fixed-delay="5000"/>
</int-jpa:inbound-channel-adapter>

<int-jpa:outbound-channel-adapter
    entity-manager-factory="entityManagerFactory"
    entity-class="com.example.Student"
    channel="inputChannel"/>

<int-jpa:retrieving-outbound-gateway
    request-channel="requestChannel"
    reply-channel="replyChannel"
    entity-manager-factory="entityManagerFactory"
    jpa-query="SELECT s FROM Student s WHERE s.id = :id"/>

<int-jpa:updating-outbound-gateway
    request-channel="requestChannel"
    reply-channel="replyChannel"
    entity-manager-factory="entityManagerFactory"
    entity-class="com.example.Student"
    persist-mode="MERGE"/>

XML Configuration

Types

PersistMode

enum PersistMode {
    PERSIST,  // Use EntityManager.persist()
    MERGE,    // Use EntityManager.merge()
    DELETE    // Use EntityManager.remove()
}

OutboundGatewayType

enum OutboundGatewayType {
    UPDATING,    // Persists entities to database
    RETRIEVING   // Retrieves entities from database
}

JpaParameter

class JpaParameter {
    JpaParameter()
    JpaParameter(String name, Object value, String expression)
    JpaParameter(Object value, String expression)

    String getName()
    void setName(String name)
    Object getValue()
    void setValue(Object value)
    String getExpression()
    void setExpression(String expression)
}

JpaOperationFailedException

class JpaOperationFailedException extends RuntimeException {
    String getOffendingJPAQl()
}

JpaUtils

final class JpaUtils {
    static final String DELETE_ALL_QUERY_STRING = "delete from %s x"

    static String detectAlias(String query)
    static <T> Query applyAndBind(String queryString, Iterable<T> entities, EntityManager entityManager)
    static String getQueryString(String template, String entityName)
    static String getEntityName(EntityManager em, Class<?> entityClass)
}

Utility methods for JPA operations, including query string manipulation, alias detection, and entity name resolution.

Configuration Approaches

Spring Integration JPA supports three configuration approaches:

1. Java DSL (recommended) - Fluent API via Jpa factory class
2. Programmatic - Direct instantiation and configuration
3. XML Namespace - Spring XML configuration with int-jpa namespace

All approaches support the same capabilities and can be mixed in the same application.