Execute get books and get scopes via cradleDispatcher

README.md

@@ -1,4 +1,4 @@
-# Report data provider (5.16.0)
+# Report data provider (5.17.0)
 
 # Overview
 This component serves as a backend for rpt-viewer. It will connect to the cassandra database via cradle api and expose the data stored in there as REST resources.
@@ -192,63 +192,67 @@ metadata:
   name: rpt-data-provider
 spec:
   imageName: ghcr.io/th2-net/th2-rpt-data-provider
-  imageVersion: 5.7.0 // change this line if you want to use a newer version
+  imageVersion: 5.17.0-dev // change this line if you want to use a newer version
   type: th2-rpt-data-provider
   customConfig:
     hostname: localhost
     port: 8080
-    responseTimeout: 60000 // maximum request processing time in milliseconds
+    responseTimeout: 60000 # maximum request processing time in milliseconds
 
-    eventCacheSize: 1000 // internal event cache size
-    messageCacheSize: 1000 // internal message cache size
-    serverCacheTimeout: 60000 // cached event lifetime in milliseconds
+    eventCacheSize: 1000      # internal event cache size
+    messageCacheSize: 1000    # internal message cache size
+    serverCacheTimeout: 60000 # cached event lifetime in milliseconds
 
-    ioDispatcherThreadPoolSize: 10 // thread pool size for blocking database calls
-    codecResponseTimeout: 6000 // if a codec doesn't respond in time, requested message is returned with a 'null' body
-    checkRequestsAliveDelay: 2000 // response channel check interval in milliseconds
+    ioDispatcherThreadPoolSize: 10  # thread pool size for blocking database calls
+    codecResponseTimeout: 6000      # if a codec doesn't respond in time, requested message is returned with a 'null' body
+    checkRequestsAliveDelay: 2000   # response channel check interval in milliseconds
 
-    enableCaching: true // enables proxy and client cache (Cache-control response headers)
-    notModifiedObjectsLifetime: 3600 // max-age in seconds
-    rarelyModifiedObjects: 500 // max-age in seconds
-
-    sseEventSearchStep: 200 // step size in seconds when requesting events 
-    keepAliveTimeout: 5000 // timeout in milliseconds. keep_alive sending frequency
-    cradleDispatcherPoolSize: 1 // number of threads in the cradle dispatcher
+    enableCaching: true               # enables proxy and client cache (Cache-control response headers)
+    notModifiedObjectsLifetime: 3600  # max-age in seconds
+    rarelyModifiedObjects: 500        # max-age in seconds
+
+    limitForParentMode: 'defailt' # mode of approach to store event id to children number. Supported: default, hash values
+
+    sseEventSearchStep: 200       # step size in seconds when requesting events 
+    keepAliveTimeout: 5000        # timeout in milliseconds. keep_alive sending frequency
+    cradleDispatcherPoolSize: 1   # number of threads in the cradle dispatcher
+    requestDispatcherPoolSize: 1  # number of threads in the request dispatcher
+    responseDispatcherPoolSize: 1 # number of threads in the response dispatcher
 
-    messageExtractorOutputBatchBuffer: 1       // buffer size of message search pipeline
+    messageExtractorOutputBatchBuffer: 1 # buffer size of message search pipeline
     messageConverterOutputBatchBuffer: 1
     messageDecoderOutputBatchBuffer: 1
     messageUnpackerOutputMessageBuffer: 100
     messageFilterOutputMessageBuffer: 100
     messageMergerOutputMessageBuffer: 10
 
-    messageIdsLookupLimitDays: 7            // lookup limit value for seacing next and previous message ids.  
+    messageIdsLookupLimitDays: 7 # lookup limit value for seacing next and previous message ids.  
 
-    codecPendingBatchLimit: 16              // the total number of messages sent to the codec batches in parallel for all pipelines
-    codecCallbackThreadPool: 4              // thread pool for parsing messages received from codecs
-    codecRequestThreadPool: 1               // thread pool for sending message to codecs
-    grpcWriterMessageBuffer: 10            // buffer before send grpc response
+    codecPendingBatchLimit: 16  # the total number of messages sent to the codec batches in parallel for all pipelines
+    codecCallbackThreadPool: 4  # thread pool for parsing messages received from codecs
+    codecRequestThreadPool: 1   # thread pool for sending message to codecs
+    grpcWriterMessageBuffer: 10 # buffer before send grpc response
 
-    sendEmptyDelay: 100 // frequency of sending empty messages
+    sendEmptyDelay: 100 # frequency of sending empty messages
 
-    eventSearchChunkSize: 64 // the size of event chunks during sse search and the maximum size of the batch of messages upon request getEvents
+    eventSearchChunkSize: 64 # the size of event chunks during sse search and the maximum size of the batch of messages upon request getEvents
 
-    grpcThreadPoolSize: 20 // thread pool size for grpc requests
+    grpcThreadPoolSize: 20 # thread pool size for grpc requests
 
-    useStrictMode: false // if true throw an exception when bad messages are received from the codec otherwise return messages with null body and type
+    useStrictMode: false # if true throw an exception when bad messages are received from the codec otherwise return messages with null body and type
 
-    serverType: HTTP // provider server type. Allows 'HTTP' and 'GRPC' (case sensetive).
+    serverType: HTTP # provider server type. Allows 'HTTP' and 'GRPC' (case sensetive).
 
-    codecUsePinAttributes: true // send raw message to specified codec (true) or send to all codecs (false)
+    codecUsePinAttributes: true # send raw message to specified codec (true) or send to all codecs (false)
 
-    eventSearchTimeOffset: 5000 // sets the offset in milliseconds on search events. (startTimestamp - eventSearchTimeOffset) and (endTimestamp + eventSearchTimeOffset)     
+    eventSearchTimeOffset: 5000 # sets the offset in milliseconds on search events. (startTimestamp - eventSearchTimeOffset) and (endTimestamp + eventSearchTimeOffset)     
 
-    searchBySessionGroup: true // if true data-provider uses the session alias to group cache and translates http / gRPC requests by session alias to group th2 storage request 
-    aliasToGroupCacheSize: 1000 // the size of cache for the mapping between session alias and group.
+    searchBySessionGroup: true # if true data-provider uses the session alias to group cache and translates http / gRPC requests by session alias to group th2 storage request 
+    aliasToGroupCacheSize: 1000 # the size of cache for the mapping between session alias and group.
 
-    useTransportMode: true // if true data-provider uses th2 transport protocol to interact with thw codecs
+    useTransportMode: true # if true data-provider uses th2 transport protocol to interact with thw codecs
 
-  pins: // pins are used to communicate with codec components to parse message data
+  pins: # pins are used to communicate with codec components to parse message data
     mq:
       subscribers:
       - name: from_codec
@@ -296,6 +300,20 @@ spec:
 
 # Release notes
 
+## 5.17.0
+* Added `limitForParentMode` option (default value is `default`) to manage logic of holding mapping between event id and number of children during event sse query execution.<br>
+  Supported two values:
+  * `default` - event ids are stored as string. Usual event id has length 40 chas or longer.
+  * `hash` - event ids are stored as long hash calculated from origin value.
+* Added `requestDispatcherPoolSize` (default value is `1`) to manage number of threads processed event / message stream during a sse query.
+* Added `responseDispatcherPoolSize` (default value is `1`) to manage number of threads serialised event stream during a sse query.
+* Updated:
+  * cradle API: `5.6.0-dev`
+  * kotlin: `2.2.0`
+  * ktor-bom: `3.2.1`
+  * kotlin-logging: `7.0.7`
+  * fastutil: `8.5.16`
+
 ## 5.16.0
 * Migrated to ktor: `3.1.2`
 * Updated:

build.gradle

@@ -53,6 +53,9 @@ dependencies {
     implementation(platform('io.ktor:ktor-bom:3.2.1'))
     implementation 'io.ktor:ktor-server-netty'
     implementation 'io.ktor:ktor-server'
+    implementation('it.unimi.dsi:fastutil:8.5.16') {
+        because('limit parent event implementation')
+    }
 
     implementation('org.ehcache:ehcache:3.10.8') {
         exclude group: 'org.glassfish.jaxb', module: 'jaxb-runtime'

gradle.properties

@@ -1,3 +1,3 @@
 kotlin.code.style=official
-release_version=5.16.0
+release_version=5.17.0
 docker_image_name=

src/main/kotlin/com/exactpro/th2/rptdataprovider/entities/configuration/Configuration.kt

@@ -41,6 +41,7 @@ class CustomConfigurationClass {
     val sseEventSearchStep: Long = 200
 
     val keepAliveTimeout: Long = 5_000
+    val limitForParentMode: String = "default"
 
     val messageExtractorOutputBatchBuffer: Int = 1
     val messageConverterOutputBatchBuffer: Int = 1
@@ -63,6 +64,8 @@ class CustomConfigurationClass {
     val eventSearchTimeOffset = 5_000
 
     val cradleDispatcherPoolSize: Long = 1
+    val requestDispatcherPoolSize: Long = 1
+    val responseDispatcherPoolSize: Long = 1
 
     val sendEmptyDelay = 100
 
@@ -84,142 +87,144 @@ class Configuration(customConfiguration: CustomConfigurationClass) {
     }
 
     val hostname: Variable =
-        Variable("hostname", customConfiguration.hostname, "localhost")
+        Variable("hostname", customConfiguration.hostname)
 
     val port: Variable =
-        Variable("port", customConfiguration.port.toString(), "8080")
+        Variable("port", customConfiguration.port.toString())
 
     val responseTimeout: Variable =
-        Variable("responseTimeout", customConfiguration.responseTimeout.toString(), "60000")
+        Variable("responseTimeout", customConfiguration.responseTimeout.toString())
 
     val serverCacheTimeout: Variable =
-        Variable("serverCacheTimeout", customConfiguration.serverCacheTimeout.toString(), "60000")
+        Variable("serverCacheTimeout", customConfiguration.serverCacheTimeout.toString())
 
     val eventCacheSize: Variable =
-        Variable("eventCacheSize", customConfiguration.eventCacheSize.toString(), "1")
+        Variable("eventCacheSize", customConfiguration.eventCacheSize.toString())
 
     val messageCacheSize: Variable =
-        Variable("messageCacheSize", customConfiguration.messageCacheSize.toString(), "1")
+        Variable("messageCacheSize", customConfiguration.messageCacheSize.toString())
 
     val aliasToGroupCacheSize: Variable = // TODO: added value check
-        Variable("aliasToGroupCacheSize", customConfiguration.aliasToGroupCacheSize.toString(), "1000")
+        Variable("aliasToGroupCacheSize", customConfiguration.aliasToGroupCacheSize.toString())
 
     val useTransportMode: Variable =
-        Variable("useTransportMode", customConfiguration.useTransportMode.toString(), "true")
+        Variable("useTransportMode", customConfiguration.useTransportMode.toString())
 
     val ioDispatcherThreadPoolSize: Variable =
         Variable("ioDispatcherThreadPoolSize", customConfiguration.ioDispatcherThreadPoolSize.let {
             if (it < 10) logger.warn { "The optimal value of the ioDispatcherThreadPoolSize is 10. Current: $it" }
 
             it.toString()
-        }, "10")
+        })
 
     val checkRequestsAliveDelay: Variable =
-        Variable("checkRequestsAliveDelay", customConfiguration.checkRequestsAliveDelay.toString(), "2000")
+        Variable("checkRequestsAliveDelay", customConfiguration.checkRequestsAliveDelay.toString())
 
-    val enableCaching: Variable = Variable("enableCaching", customConfiguration.enableCaching.toString(), "true")
+    val enableCaching: Variable = Variable("enableCaching", customConfiguration.enableCaching.toString())
 
     val notModifiedObjectsLifetime: Variable =
-        Variable("notModifiedObjectsLifetime", customConfiguration.notModifiedObjectsLifetime.toString(), "3600")
+        Variable("notModifiedObjectsLifetime", customConfiguration.notModifiedObjectsLifetime.toString())
 
     val rarelyModifiedObjects: Variable =
-        Variable("rarelyModifiedObjects", customConfiguration.rarelyModifiedObjects.toString(), "500")
+        Variable("rarelyModifiedObjects", customConfiguration.rarelyModifiedObjects.toString())
 
     val sseEventSearchStep: Variable =
-        Variable("sseEventSearchStep", customConfiguration.sseEventSearchStep.toString(), "200")
+        Variable("sseEventSearchStep", customConfiguration.sseEventSearchStep.toString())
 
     val keepAliveTimeout: Variable =
-        Variable("keepAliveTimeout", customConfiguration.keepAliveTimeout.toString(), "5000")
+        Variable("keepAliveTimeout", customConfiguration.keepAliveTimeout.toString())
+
+    val limitForParentMode: Variable =
+        Variable("limitForParentMode", customConfiguration.limitForParentMode)
 
     val messageExtractorOutputBatchBuffer: Variable =
         Variable(
             "messageExtractorOutputBatchBuffer",
-            customConfiguration.messageExtractorOutputBatchBuffer.toString(),
-            "2"
+            customConfiguration.messageExtractorOutputBatchBuffer.toString()
         )
 
     val messageConverterOutputBatchBuffer: Variable =
         Variable(
             "messageConverterOutputBatchBuffer",
-            customConfiguration.messageConverterOutputBatchBuffer.toString(),
-            "2"
+            customConfiguration.messageConverterOutputBatchBuffer.toString()
         )
 
     val messageDecoderOutputBatchBuffer: Variable =
         Variable(
             "messageDecoderOutputBatchBuffer",
-            customConfiguration.messageDecoderOutputBatchBuffer.toString(),
-            "2"
+            customConfiguration.messageDecoderOutputBatchBuffer.toString()
         )
 
     val messageUnpackerOutputMessageBuffer: Variable =
         Variable(
             "messageUnpackerOutputMessageBuffer",
-            customConfiguration.messageUnpackerOutputMessageBuffer.toString(),
-            "1000"
+            customConfiguration.messageUnpackerOutputMessageBuffer.toString()
         )
 
     val messageFilterOutputMessageBuffer: Variable =
         Variable(
             "messageFilterOutputMessageBuffer",
-            customConfiguration.messageFilterOutputMessageBuffer.toString(),
-            "1000"
+            customConfiguration.messageFilterOutputMessageBuffer.toString()
         )
 
     val messageMergerOutputMessageBuffer: Variable =
         Variable(
             "messageMergerOutputMessageBuffer",
-            customConfiguration.messageMergerOutputMessageBuffer.toString(),
-            "10"
+            customConfiguration.messageMergerOutputMessageBuffer.toString()
         )
 
     val messageIdsLookupLimitDays: Variable = Variable(
         "messageIdsLookupLimitDays",
-        customConfiguration.messageIdsLookupLimitDays.toString(),
-        "7"
+        customConfiguration.messageIdsLookupLimitDays.toString()
     )
 
     val codecResponseTimeout: Variable = Variable(
         "codecResponseTimeout",
-        customConfiguration.codecResponseTimeout.toString(), "6000"
+        customConfiguration.codecResponseTimeout.toString()
     )
 
     val codecPendingBatchLimit: Variable =
-        Variable("codecPendingBatchLimit", customConfiguration.codecPendingBatchLimit.toString(), "200")
+        Variable("codecPendingBatchLimit", customConfiguration.codecPendingBatchLimit.toString())
 
     val codecCallbackThreadPool: Variable =
-        Variable("codecCallbackThreadPool", customConfiguration.codecCallbackThreadPool.toString(), "10")
+        Variable("codecCallbackThreadPool", customConfiguration.codecCallbackThreadPool.toString())
 
     val codecRequestThreadPool: Variable =
-        Variable("codecRequestThreadPool", customConfiguration.codecRequestThreadPool.toString(), "1")
+        Variable("codecRequestThreadPool", customConfiguration.codecRequestThreadPool.toString())
 
     val grpcWriterMessageBuffer: Variable =
-        Variable("grpcWriterMessageBuffer", customConfiguration.grpcWriterMessageBuffer.toString(), "100")
+        Variable("grpcWriterMessageBuffer", customConfiguration.grpcWriterMessageBuffer.toString())
 
     val cradleDispatcherPoolSize: Variable =
-        Variable("cradleDispatcherPoolSize", customConfiguration.cradleDispatcherPoolSize.toString(), "1")
+        Variable("cradleDispatcherPoolSize", customConfiguration.cradleDispatcherPoolSize.toString())
+
+    val requestDispatcherPoolSize: Variable =
+        Variable("requestDispatcherPoolSize", customConfiguration.requestDispatcherPoolSize.toString())
+
+    val responseDispatcherPoolSize: Variable =
+        Variable("responseDispatcherPoolSize", customConfiguration.responseDispatcherPoolSize.toString())
 
     val sendEmptyDelay: Variable =
-        Variable("sendEmptyDelay", customConfiguration.sendEmptyDelay.toString(), "100")
+        Variable("sendEmptyDelay", customConfiguration.sendEmptyDelay.toString())
 
     val eventSearchChunkSize: Variable =
-        Variable("eventSearchChunkSize", customConfiguration.eventSearchChunkSize.toString(), "64")
+        Variable("eventSearchChunkSize", customConfiguration.eventSearchChunkSize.toString())
 
     val useStrictMode: Variable =
-        Variable("useStrictMode", customConfiguration.useStrictMode.toString(), "false")
+        Variable("useStrictMode", customConfiguration.useStrictMode.toString())
 
     val serverType: Variable =
-        Variable("serverType", customConfiguration.serverType.toString(), "HTTP")
+        Variable("serverType", customConfiguration.serverType.toString())
 
     val codecUsePinAttributes: Variable =
-        Variable("codecUsePinAttributes", customConfiguration.codecUsePinAttributes.toString(), "true")
+        Variable("codecUsePinAttributes", customConfiguration.codecUsePinAttributes.toString())
 
     val grpcThreadPoolSize: Variable =
-        Variable("grpcThreadPoolSize", customConfiguration.grpcThreadPoolSize.toString(), "20")
+        Variable("grpcThreadPoolSize", customConfiguration.grpcThreadPoolSize.toString())
 
     val eventSearchTimeOffset: Variable =
-        Variable("eventSearchTimeOffset", customConfiguration.eventSearchTimeOffset.toString(), "5000")
+        Variable("eventSearchTimeOffset", customConfiguration.eventSearchTimeOffset.toString())
 
     val eventSearchGap: Variable =
-        Variable("eventSearchGap", customConfiguration.eventSearchGap.toString(), "60")
+        Variable("eventSearchGap", customConfiguration.eventSearchGap.toString())
 }

src/main/kotlin/com/exactpro/th2/rptdataprovider/entities/configuration/Variable.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2024 Exactpro (Exactpro Systems Limited)
+ * Copyright 2020-2025 Exactpro (Exactpro Systems Limited)
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -21,22 +21,17 @@ import io.github.oshai.kotlinlogging.KotlinLogging
 
 class Variable(
     name: String,
-    param: String?,
-    defaultValue: String,
+    param: String,
     showInLog: Boolean = true
 ) {
     private val logger = KotlinLogging.logger { }
 
     val value: String = param
         .also {
             logger.info {
-                val valueToLog = if (showInLog) it ?: defaultValue else "*****"
+                val valueToLog = if (showInLog) it else "*****"
 
-                if (it == null)
-                    "property '$name' is not set - defaulting to '$valueToLog'"
-                else
-                    "property '$name' is set to '$valueToLog'"
+                "property '$name' is set to '$valueToLog'"
             }
         }
-        ?: defaultValue
 }