Fix HibernateOrmStateStore

quarkusio · Nov 23, 2022 · 896e85f · 896e85f
1 parent d16119e
commit 896e85f
Show file tree

Hide file tree

Showing 3 changed files with 19 additions and 7 deletions.
diff --git a/...ime/src/main/java/io/quarkus/smallrye/reactivemessaging/kafka/HibernateOrmStateStore.java b/...ime/src/main/java/io/quarkus/smallrye/reactivemessaging/kafka/HibernateOrmStateStore.java
@@ -72,19 +72,29 @@ public Uni<Map<TopicPartition, ProcessingState<?>>> fetchProcessingState(Collect
             Object[] ids = partitions.stream()
                     .map(tp -> new CheckpointEntityId(consumerGroupId, tp))
                     .toArray(Object[]::new);
-            return Vertx.currentContext().executeBlocking(Uni.createFrom().item(() -> {
+            return Vertx.currentContext().executeBlocking(Uni.createFrom().emitter(emitter -> {
                 List<CheckpointEntity> fetched = new ArrayList<>();
+                Transaction tx = null;
                 try (Session session = sf.openSession()) {
+                    tx = session.beginTransaction();
                     for (Object id : ids) {
                         CheckpointEntity entity = session.find(stateType, id);
                         if (entity != null) {
                             fetched.add(entity);
                         }
                     }
+                    session.flush();
+                    tx.commit();
+                    emitter.complete(fetched.stream().filter(e -> e != null && CheckpointEntity.topicPartition(e) != null)
+                            .collect(Collectors.toMap(CheckpointEntity::topicPartition,
+                                    e -> new ProcessingState<>(e, e.offset))));
+                } catch (Throwable t) {
+                    if (tx != null) {
+                        tx.rollback();
+                    }
+                    emitter.fail(t);
                 }
-                return fetched.stream().filter(e -> e != null && CheckpointEntity.topicPartition(e) != null)
-                        .collect(Collectors.toMap(CheckpointEntity::topicPartition,
-                                e -> new ProcessingState<>(e, e.offset)));
+
             }));
         });
     }

diff --git a/...sts/reactive-messaging-hibernate-orm/src/main/java/io/quarkus/it/kafka/KafkaEndpoint.java b/...sts/reactive-messaging-hibernate-orm/src/main/java/io/quarkus/it/kafka/KafkaEndpoint.java
@@ -4,6 +4,7 @@
 
 import javax.inject.Inject;
 import javax.persistence.EntityManager;
+import javax.transaction.Transactional;
 import javax.ws.rs.GET;
 import javax.ws.rs.Path;
 import javax.ws.rs.Produces;
@@ -40,6 +41,7 @@ public List<Person> getPeople() {
     @GET
     @Path("/people-state")
     @Produces(MediaType.APPLICATION_JSON)
+    @Transactional
     public PeopleState getPeopleState() {
         return entityManager.find(PeopleState.class,
                 new CheckpointEntityId("people-checkpoint", new TopicPartition("people", 0)));

diff --git a/...ts/reactive-messaging-hibernate-orm/src/main/java/io/quarkus/it/kafka/KafkaReceivers.java b/...ts/reactive-messaging-hibernate-orm/src/main/java/io/quarkus/it/kafka/KafkaReceivers.java
@@ -24,9 +24,9 @@ public class KafkaReceivers {
 
     @Incoming("fruits-in")
     @Transactional
-    public CompletionStage<Void> persist(Message<Fruit> fruit) {
-        fruit.getPayload().persist();
-        return emitter.sendMessage(fruit).subscribeAsCompletionStage();
+    public void persist(Fruit fruit) {
+        fruit.persist();
+        emitter.sendAndAwait(fruit);
     }
 
     @Incoming("people-in")