Merge branch 'release/1.1.0'

bom/bom/pom.xml

@@ -5,7 +5,7 @@
   <parent>
     <groupId>io.holunda</groupId>
     <artifactId>camunda-bpm-correlate-root</artifactId>
-    <version>1.0.1</version>
+    <version>1.1.0</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 

bom/parent/pom.xml

@@ -5,7 +5,7 @@
   <parent>
     <groupId>io.holunda</groupId>
     <artifactId>camunda-bpm-correlate-root</artifactId>
-    <version>1.0.1</version>
+    <version>1.1.0</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 

docs/user-guide/scheduled-processing.md

@@ -24,23 +24,26 @@ correlate:
   persistence: # persistence setting
     messageMaxRetries: 100 
     messageFetchPageSize: 100
+    messageBatchSize: 1
   retry:
     retryMaxBackoffMinutes: 5 
     retryBackoffBase: 2.0 
+
 
 ```
 
-| Property                         | Values              | Meaning                                         | Default |
-|----------------------------------|---------------------|-------------------------------------------------|---------|
-| batch.mode                       | `all`, `fail_first` | Batch processing mode                           | all     |
-| batch.query.pollInitialDelay     | Duration in ISO8601 | Start delay before correlation scheduler starts | PT10S   |
-| batch.query.pollInterval         | Duration in ISO8601 | Delay between correlation attempts              | PT6S    |
-| batch.cleanup.pollInitialDelay   | Duration in ISO8601 | Start delay before clean-up scheduler starts    |         |
-| batch.cleanup.pollInterval       | Duration in ISO8601 | Delay between clean-ups                         |         |
-| persistence.messageMaxRetries    | Integer             | Maximum retries before giving up correlation    | 100     |
-| persistence.messageFetchPageSize | Integer             | Paging size by message fetch                    | 100     |
-| retry.retryMaxBackoffMinutes     | Integer             | Maximum backoff-time  in minutes                | 180     |
-| retry.retryBackoffBase           | Float               | Base for exponential backoff-time               | 180     |
+| Property                         | Values              | Meaning                                             | Default |
+|----------------------------------|---------------------|-----------------------------------------------------|---------|
+| batch.mode                       | `all`, `fail_first` | Batch processing mode                               | all     |
+| batch.query.pollInitialDelay     | Duration in ISO8601 | Start delay before correlation scheduler starts     | PT10S   |
+| batch.query.pollInterval         | Duration in ISO8601 | Delay between correlation attempts                  | PT6S    |
+| batch.cleanup.pollInitialDelay   | Duration in ISO8601 | Start delay before clean-up scheduler starts        |         |
+| batch.cleanup.pollInterval       | Duration in ISO8601 | Delay between clean-ups                             |         |
+| persistence.messageMaxRetries    | Integer             | Maximum retries before giving up correlation        | 100     |
+| persistence.messageFetchPageSize | Integer             | Paging size by message fetch                        | 100     |
+| persistence.messageBatchSize     | Integer             | Limit the number of messages processed from a batch | -1      |
+| retry.retryMaxBackoffMinutes     | Integer             | Maximum backoff-time  in minutes                    | 180     |
+| retry.retryBackoffBase           | Float               | Base for exponential backoff-time                   | 180     |
 
 ### Reading message
 
@@ -57,6 +60,13 @@ Batches of messages are checked to fulfill the following criteria:
 Messages of one batch are correlated in order of their sorting. If a correlation error occurs, the batch correlation is
 either interrupted (`fail_first` mode) or the batch is correlated to the end (`all` mode).
 
+An important parameter for batch processing is the `message-batch-size`. This parameter specifies the number of messages
+taken from a batch for synchronous correlation. Effectively, this parameter has two interesting values. Set this parameter
+to `-1` (default) and all messages from one batch will be correlated directly one after another. Set this parameter to
+`1` and the batch will be constructed, but only the first message will be correlated in current run. If successful, the 
+next message will be fetched during the next message query (after the `batch.query.pollInterval`, which should be a small interval).
+By doing so, you can deal with asynchronous continuations in your process.
+
 ### Error detection
 
 If the error is detected during the correlation, it is handled by the library. If the message time-to-live is set and the

example/axon/flight-axon/pom.xml

@@ -5,7 +5,7 @@
   <parent>
     <groupId>io.holunda</groupId>
     <artifactId>camunda-bpm-correlate-example-parent</artifactId>
-    <version>1.0.1</version>
+    <version>1.1.0</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 

example/axon/hotel-axon/pom.xml

@@ -5,7 +5,7 @@
   <parent>
     <groupId>io.holunda</groupId>
     <artifactId>camunda-bpm-correlate-example-parent</artifactId>
-    <version>1.0.1</version>
+    <version>1.1.0</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 

example/axon/reservation-axon/pom.xml

@@ -5,7 +5,7 @@
   <parent>
     <groupId>io.holunda</groupId>
     <artifactId>camunda-bpm-correlate-example-parent</artifactId>
-    <version>1.0.1</version>
+    <version>1.1.0</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 

example/common/pom.xml

@@ -5,7 +5,7 @@
   <parent>
     <groupId>io.holunda</groupId>
     <artifactId>camunda-bpm-correlate-example-parent</artifactId>
-    <version>1.0.1</version>
+    <version>1.1.0</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 

example/itest/pom.xml

@@ -5,7 +5,7 @@
   <parent>
     <groupId>io.holunda</groupId>
     <artifactId>camunda-bpm-correlate-example-parent</artifactId>
-    <version>1.0.1</version>
+    <version>1.1.0</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 

example/pom.xml

@@ -5,7 +5,7 @@
   <parent>
     <groupId>io.holunda</groupId>
     <artifactId>camunda-bpm-correlate-parent</artifactId>
-    <version>1.0.1</version>
+    <version>1.1.0</version>
     <!-- It is ok to use the extension parent, there is only Kotlin and some managed deps defined there -->
     <relativePath>../bom/parent/pom.xml</relativePath>
   </parent>

example/spring-cloud/example.sh

@@ -9,9 +9,9 @@ CLIENT_BIN=kcat
 
 case "$1" in
   "reservation")
-    JSON=$(jq ".reservationId =\""$RESERVATION_ID\"" "$DIR/../common/local/reservation-created.json")
+    JSON=$(jq ".reservationId=\"$RESERVATION_ID\"" $DIR/../common/local/reservation-created.json)
     EVENT_TYPE=io.holunda.camunda.bpm.example.common.domain.ReservationReceivedEvent
-    echo "Sending create reservation message: $JSON"
+    echo "Sending create reservation message: '$JSON'"
   ;;
 
   "flight")
@@ -28,7 +28,7 @@ case "$1" in
   "show")
     "$CLIENT_BIN" -b "$KAFKA_BOOTSTRAP_SERVER_HOST:$KAFKA_BOOTSTRAP_SERVER_PORT" \
     -t "$KAFKA_TOPIC_CORRELATE_INGRESS" \
-    -C
+    -C -f 'Key: %k, payload: %s\n'
     exit 0
   ;;
   "all")
@@ -43,6 +43,7 @@ case "$1" in
 esac
 
 echo "$JSON" | "$CLIENT_BIN" \
+  -D "\0" \
   -b "$KAFKA_BOOTSTRAP_SERVER_HOST:$KAFKA_BOOTSTRAP_SERVER_PORT" \
   -t "$KAFKA_TOPIC_CORRELATE_INGRESS" \
   -P \

example/spring-cloud/flight-kafka/pom.xml

@@ -5,7 +5,7 @@
   <parent>
     <groupId>io.holunda</groupId>
     <artifactId>camunda-bpm-correlate-example-parent</artifactId>
-    <version>1.0.1</version>
+    <version>1.1.0</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 

example/spring-cloud/hotel-kafka/pom.xml

@@ -5,7 +5,7 @@
   <parent>
     <groupId>io.holunda</groupId>
     <artifactId>camunda-bpm-correlate-example-parent</artifactId>
-    <version>1.0.1</version>
+    <version>1.1.0</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 

example/spring-cloud/reservation-kafka/pom.xml

@@ -5,7 +5,7 @@
   <parent>
     <groupId>io.holunda</groupId>
     <artifactId>camunda-bpm-correlate-example-parent</artifactId>
-    <version>1.0.1</version>
+    <version>1.1.0</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 

example/spring-cloud/reservation-kafka/src/main/kotlin/io/holunda/camunda/bpm/example/kafka/correlation/correlate/ReservationProcessingCorrelation.kt

@@ -2,8 +2,10 @@ package io.holunda.camunda.bpm.example.kafka.correlation.correlate
 
 import io.holunda.camunda.bpm.correlate.correlation.CorrelationMessage
 import io.holunda.camunda.bpm.correlate.correlation.SingleMessageCorrelationStrategy
+import io.holunda.camunda.bpm.correlate.correlation.impl.IdentityMessageComparator
 import io.holunda.camunda.bpm.correlate.correlation.impl.MessageIdCorrelationMessageComparator
 import io.holunda.camunda.bpm.correlate.event.CorrelationHint
+import io.holunda.camunda.bpm.correlate.persist.impl.DefaultMessagePersistenceService
 import io.holunda.camunda.bpm.example.common.domain.ReservationReceivedEvent
 import io.holunda.camunda.bpm.example.common.domain.flight.FlightReservationConfirmedEvent
 import io.holunda.camunda.bpm.example.common.domain.hotel.HotelReservationConfirmedEvent
@@ -52,10 +54,11 @@ class ReservationProcessingCorrelation(
             RESERVATION_ID.name to payload.bookingReference
           )
         )
+        is DefaultMessagePersistenceService.PayloadNotAvailable -> throw IllegalArgumentException("No payload was available, could not determine correlation hint for message ${message.messageMetaData}")
         else -> throw IllegalArgumentException("Could not determine correlation hint for message ${message.messageMetaData}")
       }
     }
   }
 
-  override fun correlationMessageSorter(): Comparator<CorrelationMessage> = MessageIdCorrelationMessageComparator()
+  override fun correlationMessageSorter(): Comparator<CorrelationMessage> = IdentityMessageComparator()
 }

example/spring-cloud/reservation-kafka/src/main/resources/application-camunda-correlate.yml

@@ -27,9 +27,11 @@ correlate:
   message:
     #        timeToLiveAsString: PT10S # errors during TTL seconds after receiving are ignored
     payloadEncoding: jackson  # our bytes are actually JSON written by Jackson.
+
   persistence:
-    messageMaxRetries: 5 # default 100 -> will try to deliver 5 times at most
-    messageFetchPageSize: 100 # default 100
+    message-max-retries: 5 # default 100 -> will try to deliver 5 times at most
+    message-fetch-page-size: 100 # default 100
+    # message-batch-size: 1 # default is -1 meaning unlimited
   retry:
     retryMaxBackoffMinutes: 5 # default 180 -> maximum 5 minutes between retries
     retryBackoffBase: 5.0 # value in minutes default 2.0 -> base in the power of retry to calculate the next retry

example/spring-cloud/reservation-kafka/src/main/resources/application.yml

@@ -18,7 +18,7 @@ spring:
     stream:
       bindings:
         correlate-ingress-binding:
-          content-type: application/json
+          # content-type: application/octet-stream // do we need this?
           destination: ${KAFKA_TOPIC_CORRELATE_INGRES:correlate-ingress}
           binder: correlate-ingress-binder
           group: ${KAFKA_GROUP_ID}
@@ -34,9 +34,9 @@ spring:
           brokers: ${KAFKA_BOOTSTRAP_SERVER_HOST:localhost}:${KAFKA_BOOTSTRAP_SERVER_PORT:9092}
           configuration:
             security.protocol: ${KAFKA_SECURITY_PROTOCOL_OVERRIDE:PLAINTEXT}
-#          consumer-properties:
-#            key-deserializer: org.apache.kafka.common.serialization.ByteArrayDeserializer
-#            value-deserializer: org.apache.kafka.common.serialization.ByteArrayDeserializer
+          consumerProperties:
+            key.deserializer: org.apache.kafka.common.serialization.ByteArrayDeserializer
+            value.deserializer: org.apache.kafka.common.serialization.ByteArrayDeserializer
 
 camunda:
   bpm:

extension/axon/pom.xml

@@ -5,7 +5,7 @@
   <parent>
     <groupId>io.holunda</groupId>
     <artifactId>camunda-bpm-correlate-parent</artifactId>
-    <version>1.0.1</version>
+    <version>1.1.0</version>
     <relativePath>../../bom/parent/pom.xml</relativePath>
   </parent>
 

extension/cockpit-plugin/pom.xml

@@ -5,7 +5,7 @@
   <parent>
     <groupId>io.holunda</groupId>
     <artifactId>camunda-bpm-correlate-parent</artifactId>
-    <version>1.0.1</version>
+    <version>1.1.0</version>
     <relativePath>../../bom/parent/pom.xml</relativePath>
   </parent>
 

extension/cockpit-plugin/src/main/frontend/components/edit-retries-modal.tsx

@@ -42,7 +42,7 @@ export const EditRetriesModal = ({ message, onClose, onSubmit, maxRetries }: Edi
       <ModalBody>
         <form>
           <span>Message could not be correlated from the first attempt.</span><br />
-          {retries == maxRetries ?
+          {retries >= maxRetries ?
               <div className="form-group">
                 <span>All ${maxRetries} are exhausted. To continue retries please decrease the number of retries below:</span>
                 <label htmlFor={retriesInputId}>Retries</label>

extension/core/pom.xml

@@ -5,7 +5,7 @@
   <parent>
     <groupId>io.holunda</groupId>
     <artifactId>camunda-bpm-correlate-parent</artifactId>
-    <version>1.0.1</version>
+    <version>1.1.0</version>
     <relativePath>../../bom/parent/pom.xml</relativePath>
   </parent>
 

extension/core/src/main/kotlin/io/holunda/camunda/bpm/correlate/correlation/impl/IdentityMessageComparator.kt

@@ -0,0 +1,12 @@
+package io.holunda.camunda.bpm.correlate.correlation.impl
+
+import io.holunda.camunda.bpm.correlate.correlation.CorrelationMessage
+
+/**
+ * Correlation message comparator taking not changing the order.
+ */
+class IdentityMessageComparator : Comparator<CorrelationMessage> {
+  override fun compare(left: CorrelationMessage, right: CorrelationMessage): Int = -1
+}
+
+

extension/core/src/main/kotlin/io/holunda/camunda/bpm/correlate/persist/MessageRepository.kt

@@ -10,11 +10,11 @@ interface MessageRepository {
 
   /**
    * Finds all messages.
-   * @param page number start element to fetch from.
+   * @param page number start element to fetch from. Starts from 0.
    * @param pageSize number elements to fetch.
    * @return list of all messages.
    */
-  fun findAll(page: Int, pageSize: Int): List<MessageEntity>
+  fun findAll(page: Int = 0, pageSize: Int): List<MessageEntity>
 
   /**
    * Finds all messages without payload.

extension/core/src/main/kotlin/io/holunda/camunda/bpm/correlate/persist/impl/DefaultMessagePersistenceService.kt

@@ -96,7 +96,7 @@ class DefaultMessagePersistenceService(
     logger.debug { "Built ${batches.size} batches." }
 
     /*
-     * Fast access fun to retry info for message.
+     * Fast access function to retry info for message.
      */
     fun CorrelationMessage.retryInfo() = messagesWithRetries.getValue(this)
 
@@ -110,6 +110,12 @@ class DefaultMessagePersistenceService(
         // take those without errors or if they are due for retry
         hasNoErrors || dueForRetry
       }
+    }.map { correlationBatch ->
+      if (messagePersistenceConfig.batchSizeLimit() != -1) {
+        correlationBatch.copy(correlationMessages = correlationBatch.correlationMessages.take(messagePersistenceConfig.batchSizeLimit()))
+      } else {
+        correlationBatch
+      }
     }
   }
 

extension/core/src/main/kotlin/io/holunda/camunda/bpm/correlate/persist/impl/MessagePersistenceConfig.kt

@@ -13,5 +13,12 @@ interface MessagePersistenceConfig {
    * Retrieves the page size for paging.
    */
   fun getPageSize(): Int
+
+  /**
+   * Limits the global size of messages after batch building process. Defaults to unlimited.
+   * Special value of interest is 1, because only the first message in batch is taken in current run,
+   * and all others are ignored and will be processed during the next correlation attempt.
+   */
+  fun batchSizeLimit(): Int = -1
 }
 

extension/core/src/main/kotlin/io/holunda/camunda/bpm/correlate/persist/impl/MyBatisMessageMapper.kt

@@ -15,7 +15,7 @@ interface MyBatisMessageMapper {
   /**
    * Finds all messages using paging.
    */
-  @Select("SELECT * FROM COR_MESSAGE")
+  @Select("SELECT * FROM COR_MESSAGE M ORDER BY M.INSERTED ASC")
   @Results(
     value = [
       Result(property = "id", column = "ID", jdbcType = JdbcType.VARCHAR),
@@ -36,7 +36,7 @@ interface MyBatisMessageMapper {
   /**
    * Find all message light objects (message without payload).
    */
-  @Select("SELECT * FROM COR_MESSAGE M WHERE M.ERROR IS NOT NULL")
+  @Select("SELECT * FROM COR_MESSAGE M WHERE M.ERROR IS NOT NULL ORDER BY M.INSERTED ASC")
   @Results(
     value = [
       Result(property = "id", column = "ID", jdbcType = JdbcType.VARCHAR),
@@ -57,7 +57,7 @@ interface MyBatisMessageMapper {
   /**
    * Loads all messages.
    */
-  @Select("SELECT * FROM COR_MESSAGE")
+  @Select("SELECT * FROM COR_MESSAGE M ORDER BY M.INSERTED ASC")
   @Results(
     value = [
       Result(property = "id", column = "ID", jdbcType = JdbcType.VARCHAR),
@@ -79,7 +79,7 @@ interface MyBatisMessageMapper {
   /**
    * Finds a message by id.
    */
-  @Select("SELECT * from COR_MESSAGE M WHERE M.ID=#{id}")
+  @Select("SELECT * from COR_MESSAGE M WHERE M.ID=#{id} ORDER BY M.INSERTED ASC")
   @Results(
     value = [
       Result(property = "id", column = "ID"),