diff --git a/customer-api-provider/src/main/java/de/samples/schulung/quarkus/persistence/CustomersSinkJdbcImpl.java b/customer-api-provider/src/main/java/de/samples/schulung/quarkus/persistence/CustomersSinkJdbcImpl.java
new file mode 100644
index 0000000..631b027
--- /dev/null
+++ b/customer-api-provider/src/main/java/de/samples/schulung/quarkus/persistence/CustomersSinkJdbcImpl.java
@@ -0,0 +1,232 @@
+package de.samples.schulung.quarkus.persistence;
+
+import de.samples.schulung.quarkus.domain.Customer;
+import de.samples.schulung.quarkus.domain.Customer.CustomerState;
+import de.samples.schulung.quarkus.domain.CustomersSink;
+import io.quarkus.arc.properties.IfBuildProperty;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.enterprise.inject.Typed;
+import lombok.RequiredArgsConstructor;
+
+import javax.sql.DataSource;
+import java.sql.*;
+import java.time.LocalDate;
+import java.util.Collection;
+import java.util.LinkedList;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.stream.Stream;
+
+@ApplicationScoped
+@Typed(CustomersSink.class)
+@RequiredArgsConstructor
+@IfBuildProperty(
+  name = "persistence.sink.implementation",
+  stringValue = "jdbc"
+)
+public class CustomersSinkJdbcImpl implements CustomersSink {
+
+  private final DataSource ds;
+
+  /* ******************************************************* *
+   * Converter methods - could be converter objects instead  *
+   * ******************************************************* */
+
+  private static UUID convertUuid(String uuid) {
+    return Optional.ofNullable(uuid)
+      .map(UUID::fromString)
+      .orElse(null);
+  }
+
+  private static String convertUuid(UUID uuid) {
+    return Optional.ofNullable(uuid)
+      .map(UUID::toString)
+      .orElse(null);
+  }
+
+  private static LocalDate convertDate(Date date) {
+    return Optional.ofNullable(date)
+      .map(Date::toLocalDate)
+      .orElse(null);
+  }
+
+  private static Date convertDate(LocalDate date) {
+    return Optional.ofNullable(date)
+      .map(Date::valueOf)
+      .orElse(null);
+  }
+
+  private static CustomerState convertState(int value) {
+    return CustomerState.values()[value];
+  }
+
+  private static int convertState(CustomerState value) {
+    return Optional.ofNullable(value)
+      .map(CustomerState::ordinal)
+      .orElse(0);
+
+  }
+
+  /* ******************************************************* *
+   * Row Mapping - could be a RowMapper object instead       *
+   * ******************************************************* */
+
+  private static Customer readSingle(ResultSet rs) throws SQLException {
+    return Customer.builder()
+      .uuid(convertUuid(rs.getString("UUID")))
+      .birthday(convertDate(rs.getDate("BIRTH_DATE")))
+      .name(rs.getString("NAME"))
+      .state(convertState(rs.getInt("STATE")))
+      .build();
+  }
+
+  private static Stream<Customer> readAll(ResultSet rs) throws SQLException {
+    Collection<Customer> result = new LinkedList<>();
+    while (rs.next()) {
+      result.add(readSingle(rs));
+    }
+    return result.stream();
+  }
+
+  /* ******************************************************* *
+   * CustomersSink implementation                            *
+   * ******************************************************* */
+
+  @Override
+  public Stream<Customer> findAll() {
+    try (Connection con = ds.getConnection();
+         Statement stmt = con.createStatement();
+         ResultSet rs = stmt.executeQuery(
+           "select * from CUSTOMERS"
+         )) {
+
+      return readAll(rs);
+
+    } catch (SQLException e) {
+      throw new RuntimeException(e); // eigene Exception?
+    }
+  }
+
+  @Override
+  public Stream<Customer> findByState(CustomerState state) {
+    try (Connection con = ds.getConnection();
+         PreparedStatement stmt = con.prepareStatement(
+           "select * from CUSTOMERS where STATE=?"
+         )) {
+
+      stmt.setInt(1, convertState(state));
+
+      try (ResultSet rs = stmt.executeQuery()) {
+        return readAll(rs);
+      }
+
+    } catch (SQLException e) {
+      throw new RuntimeException(e); // eigene Exception?
+    }
+  }
+
+  @Override
+  public Optional<Customer> findByUuid(UUID uuid) {
+    try (Connection con = ds.getConnection();
+         PreparedStatement stmt = con.prepareStatement(
+           "select * from CUSTOMERS where UUID=?"
+         )) {
+
+      stmt.setString(1, convertUuid(uuid));
+
+      try (ResultSet rs = stmt.executeQuery()) {
+        if (!rs.next()) {
+          return Optional.empty();
+        }
+        return Optional.of(readSingle(rs));
+      }
+
+    } catch (SQLException e) {
+      throw new RuntimeException(e); // eigene Exception?
+    }
+  }
+
+  @Override
+  public void insert(Customer customer) {
+    try (Connection con = ds.getConnection();
+         PreparedStatement stmt = con.prepareStatement(
+           "insert into CUSTOMERS(NAME,BIRTH_DATE,STATE) values(?,?,?)",
+           Statement.RETURN_GENERATED_KEYS
+         )) {
+
+      stmt.setString(1, customer.getName());
+      stmt.setDate(2, convertDate(customer.getBirthday()));
+      stmt.setInt(3, convertState(customer.getState()));
+      stmt.executeUpdate();
+
+      try (ResultSet rs = stmt.getGeneratedKeys()) {
+        if (!rs.next()) {
+          throw new RuntimeException("not expected"); // bessere Exception
+        }
+        customer.setUuid(convertUuid(rs.getString(1)));
+      }
+
+    } catch (SQLException e) {
+      throw new RuntimeException(e); // eigene Exception?
+    }
+  }
+
+  @Override
+  public void update(Customer customer) {
+    try (Connection con = ds.getConnection();
+         PreparedStatement stmt = con.prepareStatement(
+           "update CUSTOMERS set NAME=?, BIRTH_DATE=?, STATE=?) where UUID=?",
+           Statement.RETURN_GENERATED_KEYS
+         )) {
+
+      stmt.setString(1, customer.getName());
+      stmt.setDate(2, convertDate(customer.getBirthday()));
+      stmt.setInt(3, convertState(customer.getState()));
+      stmt.setString(4, convertUuid(customer.getUuid()));
+      stmt.executeUpdate();
+
+      try (ResultSet rs = stmt.getGeneratedKeys()) {
+        if (!rs.next()) {
+          throw new RuntimeException("not expected"); // bessere Exception
+        }
+        customer.setUuid(convertUuid(rs.getString(1)));
+      }
+
+    } catch (SQLException e) {
+      throw new RuntimeException(e); // eigene Exception?
+    }
+  }
+
+  @Override
+  public boolean delete(UUID uuid) {
+    try (Connection con = ds.getConnection();
+         PreparedStatement stmt = con.prepareStatement(
+           "delete from CUSTOMERS where UUID=?"
+         )) {
+
+      stmt.setString(1, convertUuid(uuid));
+      return stmt.executeUpdate() > 0;
+
+    } catch (SQLException e) {
+      throw new RuntimeException(e); // eigene Exception?
+    }
+  }
+
+  @Override
+  public long count() {
+    try (Connection con = ds.getConnection();
+         Statement stmt = con.createStatement();
+         ResultSet rs = stmt.executeQuery(
+           "select count(uuid) from CUSTOMERS"
+         )) {
+
+      if (!rs.next()) {
+        return 0;
+      }
+      return rs.getLong(1);
+
+    } catch (SQLException e) {
+      throw new RuntimeException(e); // eigene Exception?
+    }
+  }
+}
diff --git a/customer-api-provider/src/main/java/de/samples/schulung/quarkus/persistence/CustomersSinkJpaImpl.java b/customer-api-provider/src/main/java/de/samples/schulung/quarkus/persistence/CustomersSinkJpaImpl.java
new file mode 100644
index 0000000..bf10fac
--- /dev/null
+++ b/customer-api-provider/src/main/java/de/samples/schulung/quarkus/persistence/CustomersSinkJpaImpl.java
@@ -0,0 +1,95 @@
+package de.samples.schulung.quarkus.persistence;
+
+import de.samples.schulung.quarkus.domain.Customer;
+import de.samples.schulung.quarkus.domain.CustomersSink;
+import io.quarkus.arc.properties.IfBuildProperty;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.enterprise.inject.Typed;
+import jakarta.persistence.EntityManager;
+import jakarta.transaction.Transactional;
+import lombok.RequiredArgsConstructor;
+
+import java.util.Optional;
+import java.util.UUID;
+import java.util.stream.Stream;
+
+@ApplicationScoped
+@Typed(CustomersSink.class)
+@RequiredArgsConstructor
+@IfBuildProperty(
+  name = "persistence.sink.implementation",
+  stringValue = "jpa"
+)
+public class CustomersSinkJpaImpl implements CustomersSink {
+
+  private final CustomerEntityMapper mapper;
+  private final EntityManager em;
+
+  @Override
+  public Stream<Customer> findAll() {
+    return em.createQuery(
+        "select c from Customer c",
+        CustomerEntity.class
+      )
+      .getResultList()
+      .stream()
+      .map(mapper::map);
+  }
+
+  @Override
+  public Stream<Customer> findByState(Customer.CustomerState state) {
+    return em.createQuery(
+        "select c from Customer c where c.state = :state",
+        CustomerEntity.class
+      )
+      .setParameter("state", state)
+      .getResultList()
+      .stream()
+      .map(mapper::map);
+  }
+
+  @Override
+  public Optional<Customer> findByUuid(UUID uuid) {
+    return Optional
+      .ofNullable(em.find(CustomerEntity.class, uuid))
+      .map(mapper::map);
+  }
+
+  @Transactional
+  @Override
+  public void insert(Customer customer) {
+    var entity = this.mapper.map(customer);
+    em.persist(entity);
+    mapper.copy(entity, customer);
+  }
+
+  @Transactional
+  @Override
+  public void update(Customer customer) {
+    var entity = this.mapper.map(customer);
+    em.persist(entity);
+    mapper.copy(entity, customer);
+  }
+
+  @Transactional
+  @Override
+  public boolean delete(UUID uuid) {
+    var found = em.find(CustomerEntity.class, uuid);
+    if (found == null) {
+      return false;
+    }
+    em.remove(found);
+    return true;
+  }
+
+  @Override
+  public boolean existsByUuid(UUID uuid) {
+    var found = em.find(CustomerEntity.class, uuid);
+    return found != null;
+  }
+
+  @Override
+  public long count() {
+    return 0; // TODO ??
+  }
+}
diff --git a/customer-api-provider/src/main/java/de/samples/schulung/quarkus/persistence/CustomersSinkPanacheImpl.java b/customer-api-provider/src/main/java/de/samples/schulung/quarkus/persistence/CustomersSinkPanacheImpl.java
index f9d07fd..8447822 100644
--- a/customer-api-provider/src/main/java/de/samples/schulung/quarkus/persistence/CustomersSinkPanacheImpl.java
+++ b/customer-api-provider/src/main/java/de/samples/schulung/quarkus/persistence/CustomersSinkPanacheImpl.java
@@ -5,6 +5,7 @@
 import io.quarkus.arc.properties.IfBuildProperty;
 import jakarta.enterprise.context.ApplicationScoped;
 import jakarta.enterprise.inject.Typed;
+import jakarta.transaction.Transactional;
 import lombok.RequiredArgsConstructor;
 
 import java.util.Optional;
@@ -12,67 +13,70 @@
 import java.util.stream.Stream;
 
 @IfBuildProperty(
-  name = "persistence.sink.implementation",
-  stringValue = "panache"
+        name = "persistence.sink.implementation",
+        stringValue = "panache"
 )
 @ApplicationScoped
 @Typed(CustomersSink.class)
 @RequiredArgsConstructor
 public class CustomersSinkPanacheImpl implements CustomersSink {
 
-  private final CustomerEntityRepository repo;
-  private final CustomerEntityMapper mapper;
+    private final CustomerEntityRepository repo;
+    private final CustomerEntityMapper mapper;
 
-  @Override
-  public Stream<Customer> findAll() {
-    return repo.findAll()
-      .stream()
-      .map(mapper::map);
-  }
+    @Override
+    public Stream<Customer> findAll() {
+        return repo.findAll()
+                .stream()
+                .map(mapper::map);
+    }
 
-  @Override
-  public long count() {
-    return repo.count();
-  }
+    @Override
+    public long count() {
+        return repo.count();
+    }
 
-  @Override
-  public Stream<Customer> findByState(Customer.CustomerState state) {
-    return repo.list("state", state)
-      .stream()
-      .map(mapper::map);
-  }
+    @Override
+    public Stream<Customer> findByState(Customer.CustomerState state) {
+        return repo.list("state", state)
+                .stream()
+                .map(mapper::map);
+    }
 
-  @Override
-  public Optional<Customer> findByUuid(UUID uuid) {
-    return repo.findByIdOptional(uuid)
-      .map(mapper::map);
-  }
+    @Override
+    public Optional<Customer> findByUuid(UUID uuid) {
+        return repo.findByIdOptional(uuid)
+                .map(mapper::map);
+    }
 
-  @Override
-  public boolean existsByUuid(UUID uuid) {
-    return repo.findByIdOptional(uuid)
-      .isPresent();
-  }
+    @Override
+    public boolean existsByUuid(UUID uuid) {
+        return repo.findByIdOptional(uuid)
+                .isPresent();
+    }
 
-  @Override
-  public void insert(Customer customer) {
-    var entity = mapper.map(customer);
-    repo.persist(entity);
-    //customer.setUuid(entity.getUuid());
-    mapper.copy(entity, customer);
-  }
+    @Transactional
+    @Override
+    public void insert(Customer customer) {
+        var entity = mapper.map(customer);
+        repo.persist(entity);
+        //customer.setUuid(entity.getUuid());
+        mapper.copy(entity, customer);
+    }
 
-  @Override
-  public void update(Customer customer) {
-    var entity = mapper.map(customer);
-    repo.persist(entity);
-    //customer.setUuid(entity.getUuid());
-    mapper.copy(entity, customer);
-  }
+    @Transactional
+    @Override
+    public void update(Customer customer) {
+        var entity = mapper.map(customer);
+        repo.persist(entity);
+        //customer.setUuid(entity.getUuid());
+        mapper.copy(entity, customer);
+    }
 
-  @Override
-  public boolean delete(UUID uuid) {
-    return repo.deleteById(uuid);
-  }
+    @Transactional
+    @Override
+    public boolean delete(UUID uuid) {
+        return repo.deleteById(uuid);
+    }
 
 }
diff --git a/customer-api-provider/src/main/resources/application.properties b/customer-api-provider/src/main/resources/application.properties
index 6d2de1f..75b4fdc 100644
--- a/customer-api-provider/src/main/resources/application.properties
+++ b/customer-api-provider/src/main/resources/application.properties
@@ -7,3 +7,4 @@ quarkus.http.cors=true
 persistence.sink.implementation=panache
 %test.persistence.sink.implementation=in-memory
 quarkus.hibernate-orm.log.sql=true
+%dev.quarkus.hibernate-orm.database.generation=update
diff --git a/customer-api-provider/src/test/java/de/samples/schulung/quarkus/persistence/PersistenceJdbcCustomersSinkTests.java b/customer-api-provider/src/test/java/de/samples/schulung/quarkus/persistence/PersistenceJdbcCustomersSinkTests.java
new file mode 100644
index 0000000..8b22c4c
--- /dev/null
+++ b/customer-api-provider/src/test/java/de/samples/schulung/quarkus/persistence/PersistenceJdbcCustomersSinkTests.java
@@ -0,0 +1,29 @@
+package de.samples.schulung.quarkus.persistence;
+
+import de.samples.schulung.quarkus.domain.Customer.CustomerState;
+import de.samples.schulung.quarkus.domain.CustomersSink;
+import io.quarkus.test.TestTransaction;
+import io.quarkus.test.junit.QuarkusTest;
+import io.quarkus.test.junit.TestProfile;
+import jakarta.inject.Inject;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@QuarkusTest
+@TestTransaction
+@TestProfile(UseJdbcImplementation.class)
+public class PersistenceJdbcCustomersSinkTests {
+
+  @Inject
+  CustomersSink sink;
+
+  @DisplayName("[DB - JDBC] Simple Find By State")
+  @Test
+  void given_whenFindByState_thenResultAvailable() {
+    var result = sink.findByState(CustomerState.ACTIVE);
+    assertThat(result).isNotNull();
+  }
+
+}
diff --git a/customer-api-provider/src/test/java/de/samples/schulung/quarkus/persistence/PersistenceJpaCustomersSinkTests.java b/customer-api-provider/src/test/java/de/samples/schulung/quarkus/persistence/PersistenceJpaCustomersSinkTests.java
new file mode 100644
index 0000000..22207ce
--- /dev/null
+++ b/customer-api-provider/src/test/java/de/samples/schulung/quarkus/persistence/PersistenceJpaCustomersSinkTests.java
@@ -0,0 +1,29 @@
+package de.samples.schulung.quarkus.persistence;
+
+import de.samples.schulung.quarkus.domain.Customer.CustomerState;
+import de.samples.schulung.quarkus.domain.CustomersSink;
+import io.quarkus.test.TestTransaction;
+import io.quarkus.test.junit.QuarkusTest;
+import io.quarkus.test.junit.TestProfile;
+import jakarta.inject.Inject;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@QuarkusTest
+@TestTransaction
+@TestProfile(UseJpaImplementation.class)
+public class PersistenceJpaCustomersSinkTests {
+
+  @Inject
+  CustomersSink sink;
+
+  @DisplayName("[DB - JPA] Simple Find By State")
+  @Test
+  void given_whenFindByState_thenResultAvailable() {
+    var result = sink.findByState(CustomerState.ACTIVE);
+    assertThat(result).isNotNull();
+  }
+
+}
diff --git a/customer-api-provider/src/test/java/de/samples/schulung/quarkus/persistence/PersistencePanacheCustomersSinkTests.java b/customer-api-provider/src/test/java/de/samples/schulung/quarkus/persistence/PersistencePanacheCustomersSinkTests.java
new file mode 100644
index 0000000..a3e1ccc
--- /dev/null
+++ b/customer-api-provider/src/test/java/de/samples/schulung/quarkus/persistence/PersistencePanacheCustomersSinkTests.java
@@ -0,0 +1,46 @@
+package de.samples.schulung.quarkus.persistence;
+
+import de.samples.schulung.quarkus.domain.Customer;
+import de.samples.schulung.quarkus.domain.Customer.CustomerState;
+import de.samples.schulung.quarkus.domain.CustomersSink;
+import io.quarkus.test.TestTransaction;
+import io.quarkus.test.junit.QuarkusTest;
+import io.quarkus.test.junit.TestProfile;
+import jakarta.inject.Inject;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import java.time.LocalDate;
+import java.time.Month;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@QuarkusTest
+@TestTransaction
+@TestProfile(UsePanacheImplementation.class)
+public class PersistencePanacheCustomersSinkTests {
+
+  @Inject
+  CustomersSink sink;
+
+  @DisplayName("[DB - Panache] Simple Find By State")
+  @Test
+  void given_whenFindByState_thenResultAvailable() {
+    var result = sink.findByState(CustomerState.ACTIVE);
+    assertThat(result).isNotNull();
+  }
+
+  @DisplayName("[DB - Panache] Create Customer")
+  @Test
+  void given_whenInsert_thenUuidIsGenerated() {
+    var customer = Customer
+      .builder()
+      .name("John")
+      .birthday(LocalDate.of(2000, Month.AUGUST, 20))
+      .build();
+    sink.insert(customer);
+    assertThat(customer.getUuid())
+      .isNotNull();
+  }
+
+}
diff --git a/customer-api-provider/src/test/java/de/samples/schulung/quarkus/persistence/UseJdbcImplementation.java b/customer-api-provider/src/test/java/de/samples/schulung/quarkus/persistence/UseJdbcImplementation.java
new file mode 100644
index 0000000..7eb7261
--- /dev/null
+++ b/customer-api-provider/src/test/java/de/samples/schulung/quarkus/persistence/UseJdbcImplementation.java
@@ -0,0 +1,15 @@
+package de.samples.schulung.quarkus.persistence;
+
+import io.quarkus.test.junit.QuarkusTestProfile;
+
+import java.util.Map;
+
+public class UseJdbcImplementation implements QuarkusTestProfile {
+
+  @Override
+  public Map<String, String> getConfigOverrides() {
+    return Map.of(
+      "persistence.sink.implementation", "jdbc"
+    );
+  }
+}
diff --git a/customer-api-provider/src/test/java/de/samples/schulung/quarkus/persistence/UseJpaImplementation.java b/customer-api-provider/src/test/java/de/samples/schulung/quarkus/persistence/UseJpaImplementation.java
new file mode 100644
index 0000000..aaf6841
--- /dev/null
+++ b/customer-api-provider/src/test/java/de/samples/schulung/quarkus/persistence/UseJpaImplementation.java
@@ -0,0 +1,15 @@
+package de.samples.schulung.quarkus.persistence;
+
+import io.quarkus.test.junit.QuarkusTestProfile;
+
+import java.util.Map;
+
+public class UseJpaImplementation implements QuarkusTestProfile {
+
+  @Override
+  public Map<String, String> getConfigOverrides() {
+    return Map.of(
+      "persistence.sink.implementation", "jpa"
+    );
+  }
+}
diff --git a/customer-api-provider/src/test/java/de/samples/schulung/quarkus/persistence/UsePanacheImplementation.java b/customer-api-provider/src/test/java/de/samples/schulung/quarkus/persistence/UsePanacheImplementation.java
new file mode 100644
index 0000000..92bc779
--- /dev/null
+++ b/customer-api-provider/src/test/java/de/samples/schulung/quarkus/persistence/UsePanacheImplementation.java
@@ -0,0 +1,15 @@
+package de.samples.schulung.quarkus.persistence;
+
+import io.quarkus.test.junit.QuarkusTestProfile;
+
+import java.util.Map;
+
+public class UsePanacheImplementation implements QuarkusTestProfile {
+
+  @Override
+  public Map<String, String> getConfigOverrides() {
+    return Map.of(
+      "persistence.sink.implementation", "panache"
+    );
+  }
+}