fix: event batcher does not set parent event ID (#47)

Author: Oleg Smirnov

.github/workflows/build-dev-release.yml

@@ -14,4 +14,7 @@ jobs:
       sonatypePassword: ${{ secrets.SONATYPE_NEXUS_PASSWORD }}
       sonatypeSigningKey: ${{ secrets.SONATYPE_GPG_ARMORED_KEY }}
       sonatypeSigningPassword: ${{ secrets.SONATYPE_SIGNING_PASSWORD }}
-      nvd-api-key: ${{ secrets.NVD_APIKEY }}
+      nvd-api-key: ${{ secrets.NVD_APIKEY }}
+      cisa-domain: ${{ secrets.CISA_DOMAIN }}
+      cisa-user: ${{ secrets.CISA_USER }}
+      cisa-password: ${{ secrets.CISA_PWD }}

.github/workflows/build-release.yml

@@ -14,4 +14,7 @@ jobs:
       sonatypePassword: ${{ secrets.SONATYPE_NEXUS_PASSWORD }}
       sonatypeSigningKey: ${{ secrets.SONATYPE_GPG_ARMORED_KEY }}
       sonatypeSigningPassword: ${{ secrets.SONATYPE_SIGNING_PASSWORD }}
-      nvd-api-key: ${{ secrets.NVD_APIKEY }}
+      nvd-api-key: ${{ secrets.NVD_APIKEY }}
+      cisa-domain: ${{ secrets.CISA_DOMAIN }}
+      cisa-user: ${{ secrets.CISA_USER }}
+      cisa-password: ${{ secrets.CISA_PWD }}

.github/workflows/build-sanpshot.yml

@@ -19,4 +19,7 @@ jobs:
       sonatypePassword: ${{ secrets.SONATYPE_NEXUS_PASSWORD }}
       sonatypeSigningKey: ${{ secrets.SONATYPE_GPG_ARMORED_KEY }}
       sonatypeSigningPassword: ${{ secrets.SONATYPE_SIGNING_PASSWORD }}
-      nvd-api-key: ${{ secrets.NVD_APIKEY }}
+      nvd-api-key: ${{ secrets.NVD_APIKEY }}
+      cisa-domain: ${{ secrets.CISA_DOMAIN }}
+      cisa-user: ${{ secrets.CISA_USER }}
+      cisa-password: ${{ secrets.CISA_PWD }}

README.md

@@ -1,4 +1,4 @@
-# th2 common util library (2.3.0)
+# th2 common util library (2.3.1)
 
 This is th2 java library with useful functions for developers and QA needs.
 
@@ -31,6 +31,11 @@ elapsed or number of events in it has reached `maxBatchSize` or batch size in by
 
 # Changelog
 
+## 2.3.1
+
+* fixed:
+  * `EventBatcher` does not set parent ID that leads to storing each event as an individual entity in cradle
+
 ## 2.3.0
 
 * added `RawMessageBatcher` for th2 transport protocol 

build.gradle

@@ -2,7 +2,7 @@ plugins {
     id 'maven-publish'
     id 'java-test-fixtures'
     id 'org.jetbrains.kotlin.jvm' version '1.8.22'
-    id 'com.exactpro.th2.gradle.publish' version '0.1.1'
+    id 'com.exactpro.th2.gradle.publish' version '0.2.3'
 }
 
 group = 'com.exactpro.th2'
@@ -29,10 +29,6 @@ repositories {
     }
 }
 
-kotlin {
-    jvmToolchain(11)
-}
-
 test {
     useJUnitPlatform()
 }

gradle.properties

@@ -1,3 +1,3 @@
-release_version=2.3.0
+release_version=2.3.1
 description='th2 common utils (Java)'
 vcs_url=https://github.com/th2-net/th2-common-utils-j

src/main/kotlin/com/exactpro/th2/common/utils/event/EventBatcher.kt

@@ -27,14 +27,20 @@ import java.util.concurrent.TimeUnit.MILLISECONDS
 import java.util.concurrent.locks.ReentrantLock
 import kotlin.concurrent.withLock
 
+private const val DEFAULT_MAX_BATCH_SIZE_BYTES = 1_024L * 1_024L
+
+private const val DEFAULT_MAX_BATCH_SIZE_EVENTS = 100
+
+private const val DEFAULT_MAX_FLUSH_TIME_MILLIS = 1000L
+
 /**
  * Collects and groups events by their parent-event-id and calls `onBatch` method when `maxFlushTime`
  * for a group has elapsed or number of events in it has reached `maxBatchSize`.
  */
 class EventBatcher(
-    private val maxBatchSizeInBytes: Long = 1_024 * 1_024,
-    private val maxBatchSizeInItems: Int = 100,
-    private val maxFlushTime: Long = 1000,
+    private val maxBatchSizeInBytes: Long = DEFAULT_MAX_BATCH_SIZE_BYTES,
+    private val maxBatchSizeInItems: Int = DEFAULT_MAX_BATCH_SIZE_EVENTS,
+    private val maxFlushTime: Long = DEFAULT_MAX_FLUSH_TIME_MILLIS,
     private val executor: ScheduledExecutorService,
     private val onBatch: (EventBatch) -> Unit,
 ) : AutoCloseable {
@@ -44,11 +50,13 @@ class EventBatcher(
         .build<EventID, Batch>()
         .asMap()
 
-    fun onEvent(event: Event) = batches.getOrPut(event.parentId, ::Batch).add(event)
+    fun onEvent(event: Event) = batches.getOrPut(event.parentId) { Batch(event.parentId) }.add(event)
 
     override fun close() = batches.values.forEach(Batch::close)
 
-    private inner class Batch : AutoCloseable {
+    private inner class Batch(
+        private val parentEventID: EventID,
+    ) : AutoCloseable {
         private val lock = ReentrantLock()
         private var batch = EventBatch.newBuilder()
         private var batchSizeInBytes = BATCH_LEN_CONST
@@ -73,38 +81,47 @@ class EventBatcher(
 
         private fun send() = lock.withLock<Unit> {
             if (batch.eventsCount == 0) return
+            if (propagateParentIdToBatch() && batch.eventsCount > 1) {
+                // it only makes sense to store events as batch if we have more than one event in the batch
+                batch.parentEventId = parentEventID
+            }
             batch.build().runCatching(onBatch)
             batch.clearEvents()
+            batch.clearParentEventId()
             batchSizeInBytes = BATCH_LEN_CONST
             future.cancel(false)
         }
+
+        private fun propagateParentIdToBatch(): Boolean =
+            parentEventID != EventID.getDefaultInstance()
     }
 
     companion object {
         /**
-		* 4 - magic number
-		* 1 - protocol version
-		* 4 - message sizes
-		* Collapsed constant = 9
-	    */
+         * 4 - magic number
+         * 1 - protocol version
+         * 4 - message sizes
+         * Collapsed constant = 9
+         */
         private const val BATCH_LEN_CONST = 9L
 
         /**
-        * 2 - magic number
-        * 4 + 8 = Instant (start timestamp) long (seconds) + int (nanos) - start timestamp ID
-        * 4 - id length
-        * 4 - name length
-        * 4 - type length
-        * 4 + 8 = Instant (start timestamp) long (seconds) + int (nanos) - start timestamp parent ID
-        * 4 - parent id length
-        * 4 + 8 = Instant (end timestamp) long (seconds) + int (nanos)
-        * 1 = is success
-        * 4 = body len
-        * ===
-        * 59
-        */
+         * 2 - magic number
+         * 4 + 8 = Instant (start timestamp) long (seconds) + int (nanos) - start timestamp ID
+         * 4 - id length
+         * 4 - name length
+         * 4 - type length
+         * 4 + 8 = Instant (start timestamp) long (seconds) + int (nanos) - start timestamp parent ID
+         * 4 - parent id length
+         * 4 + 8 = Instant (end timestamp) long (seconds) + int (nanos)
+         * 1 = is success
+         * 4 = body len
+         * ===
+         * 59
+         */
         private const val EVENT_RECORD_CONST = 59L
         private const val BATCH_LENGTH_IN_BATCH = 4L
+
         /**
          * This logic is copied the estore project.
          */

src/test/kotlin/com/exactpro/th2/common/utils/event/EventBatcherTest.kt

@@ -18,8 +18,13 @@ package com.exactpro.th2.common.utils.event
 
 import com.exactpro.th2.common.grpc.Event
 import com.exactpro.th2.common.grpc.EventBatch
+import com.exactpro.th2.common.grpc.EventID
+import com.exactpro.th2.common.utils.event.EventBatcher.Companion
 import com.exactpro.th2.common.utils.event.EventBatcher.Companion.calculateSizeInBytes
+import com.google.protobuf.util.Timestamps
+import org.junit.jupiter.api.Assertions
 import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertFalse
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.assertAll
 import org.mockito.kotlin.any
@@ -55,10 +60,12 @@ class EventBatcherTest {
             verify(executor, times(1)).schedule(any(), any(), any())
             verify(future, times(1)).cancel(any())
 
+            val batch = batchCaptor.firstValue
             assertAll(
-                { assertEquals(2, batchCaptor.firstValue.eventsCount) },
-                { assertEquals(EVENT_1, batchCaptor.firstValue.getEvents(0)) },
-                { assertEquals(EVENT_2, batchCaptor.firstValue.getEvents(1)) },
+                { assertEquals(2, batch.eventsCount) },
+                { assertEquals(PARENT_ID, batch.parentEventId, "batch should same parent ID as events") },
+                { assertEquals(EVENT_1, batch.getEvents(0)) },
+                { assertEquals(EVENT_2, batch.getEvents(1)) },
             )
         }
     }
@@ -89,13 +96,17 @@ class EventBatcherTest {
             verify(executor, times(3)).schedule(any(), any(), any())
             verify(future, times(2)).cancel(any())
 
+            val batch1 = batchCaptor.firstValue
             assertAll(
-                { assertEquals(1, batchCaptor.firstValue.eventsCount) },
-                { assertEquals(EVENT_1, batchCaptor.firstValue.getEvents(0)) },
+                { assertEquals(1, batch1.eventsCount) },
+                { assertFalse(batch1.hasParentEventId(), "batch with single event should not have parent ID") },
+                { assertEquals(EVENT_1, batch1.getEvents(0)) },
             )
+            val batch2 = batchCaptor.secondValue
             assertAll(
-                { assertEquals(1, batchCaptor.secondValue.eventsCount) },
-                { assertEquals(EVENT_2, batchCaptor.secondValue.getEvents(0)) },
+                { assertEquals(1, batch2.eventsCount) },
+                { assertFalse(batch2.hasParentEventId(), "batch with single event should not have parent ID") },
+                { assertEquals(EVENT_2, batch2.getEvents(0)) },
             )
         }
     }
@@ -121,28 +132,40 @@ class EventBatcherTest {
             runnableCaptor.firstValue.run()
             verify(onBatch, times(1))(any())
             verify(future, times(1)).cancel(any())
+            val batch = batchCaptor.firstValue
             assertAll(
-                { assertEquals(1, batchCaptor.firstValue.eventsCount) },
-                { assertEquals(EVENT_1, batchCaptor.firstValue.getEvents(0)) },
+                { assertEquals(1, batch.eventsCount) },
+                { assertFalse(batch.hasParentEventId(), "batch with single event should not have parent ID") },
+                { assertEquals(EVENT_1, batch.getEvents(0)) },
             )
         }
     }
 
     companion object {
+        private val PARENT_ID = EventID.newBuilder()
+            .setId("test")
+            .setScope("scope")
+            .setBookName("book")
+            .setStartTimestamp(Timestamps.now())
+            .build()
+
         private val EVENT_1 = Event.newBuilder().apply {
             idBuilder.apply {
                 id = "test_1"
             }
+            parentId = PARENT_ID
         }.build()
         private val EVENT_2 = Event.newBuilder().apply {
             idBuilder.apply {
                 id = "test_2"
             }
+            parentId = PARENT_ID
         }.build()
         private val EVENT_3 = Event.newBuilder().apply {
             idBuilder.apply {
                 id = "test_3"
             }
+            parentId = PARENT_ID
         }.build()
 
         private val EVENT_SIZE_IN_BYTES = maxOf(