R3SOL-368 Move ledger flow code from components into a new library (#…

…6353) Put ledger flow code into a new library. Uncoupled the new libraries dependencies and transitive dependencies from the avro schema. Pulling out code where needed into new modules (if there isn't a sensible place). Co-authored-by: Dan Newton <[email protected]>
corda · Oct 24, 2024 · c80cfb8 · c80cfb8
1 parent c570ab5
commit c80cfb8
Show file tree

Hide file tree

Showing 282 changed files with 1,323 additions and 739 deletions.
diff --git a/applications/workers/workers-smoketest/build.gradle b/applications/workers/workers-smoketest/build.gradle
@@ -70,6 +70,7 @@ dependencies {
 
     smokeTestImplementation project(':components:flow:flow-rest-resource-service')
     smokeTestImplementation project(':libs:crypto:crypto-core')
+    smokeTestImplementation project(':libs:crypto:crypto-core-avro')
     // TODO: this reference is only to get the serialialiser. I don't think this should be in the
     //  messagebus project.
     smokeTestImplementation project(":libs:messaging:kafka-message-bus-impl")

diff --git a/...mokeTest/kotlin/net/corda/applications/workers/smoketest/services/CryptoRestSmokeTests.kt b/...mokeTest/kotlin/net/corda/applications/workers/smoketest/services/CryptoRestSmokeTests.kt
@@ -2,7 +2,7 @@ package net.corda.applications.workers.smoketest.services
 
 import net.corda.applications.workers.smoketest.utils.PLATFORM_VERSION
 import net.corda.crypto.core.SecureHashImpl
-import net.corda.crypto.core.toAvro
+import net.corda.crypto.core.avro.toAvro
 import net.corda.data.KeyValuePair
 import net.corda.data.KeyValuePairList
 import net.corda.data.crypto.SecureHashes

diff --git a/components/chunking/chunk-db-write-impl/build.gradle b/components/chunking/chunk-db-write-impl/build.gradle
@@ -29,6 +29,7 @@ dependencies {
     implementation project(':libs:chunking:chunking-core')
     implementation project(':libs:configuration:configuration-core')
     implementation project(':libs:crypto:crypto-core')
+    implementation project(':libs:crypto:crypto-core-avro')
     implementation project(':libs:db:db-core')
     implementation project(':libs:db:db-orm')
     implementation project(':libs:db:db-admin')

diff --git a/...c/integrationTest/kotlin/net/corda/chunking/db/impl/tests/DatabaseChunkPersistenceTest.kt b/...c/integrationTest/kotlin/net/corda/chunking/db/impl/tests/DatabaseChunkPersistenceTest.kt
@@ -18,7 +18,7 @@ import net.corda.chunking.datamodel.ChunkingEntities
 import net.corda.chunking.db.impl.AllChunksReceived
 import net.corda.chunking.db.impl.persistence.database.DatabaseChunkPersistence
 import net.corda.crypto.core.SecureHashImpl
-import net.corda.crypto.core.toAvro
+import net.corda.crypto.core.avro.toAvro
 import net.corda.data.chunking.Chunk
 import net.corda.db.admin.impl.ClassloaderChangeLog
 import net.corda.db.admin.impl.LiquibaseSchemaMigratorImpl

diff --git a/...k-db-write-impl/src/main/kotlin/net/corda/chunking/db/impl/persistence/StatusPublisher.kt b/...k-db-write-impl/src/main/kotlin/net/corda/chunking/db/impl/persistence/StatusPublisher.kt
@@ -1,6 +1,6 @@
 package net.corda.chunking.db.impl.persistence
 
-import net.corda.crypto.core.toAvro
+import net.corda.crypto.core.avro.toAvro
 import net.corda.data.ExceptionEnvelope
 import net.corda.data.chunking.UploadStatus
 import net.corda.data.chunking.UploadStatusKey

diff --git a/...c/main/kotlin/net/corda/chunking/db/impl/persistence/database/DatabaseChunkPersistence.kt b/...c/main/kotlin/net/corda/chunking/db/impl/persistence/database/DatabaseChunkPersistence.kt
@@ -6,10 +6,10 @@ import net.corda.chunking.datamodel.ChunkEntity
 import net.corda.chunking.datamodel.ChunkPropertyEntity
 import net.corda.chunking.db.impl.AllChunksReceived
 import net.corda.chunking.db.impl.persistence.ChunkPersistence
+import net.corda.crypto.core.avro.toAvro
+import net.corda.crypto.core.avro.toCorda
 import net.corda.crypto.core.bytes
 import net.corda.crypto.core.parseSecureHash
-import net.corda.crypto.core.toAvro
-import net.corda.crypto.core.toCorda
 import net.corda.data.KeyValuePair
 import net.corda.data.KeyValuePairList
 import net.corda.data.chunking.Chunk

diff --git a/...rite-impl/src/test/kotlin/net/corda/chunking/db/impl/ChunkWriteToDbProcessorSimpleTest.kt b/...rite-impl/src/test/kotlin/net/corda/chunking/db/impl/ChunkWriteToDbProcessorSimpleTest.kt
@@ -4,8 +4,8 @@ import net.corda.chunking.RequestId
 import net.corda.chunking.db.impl.persistence.ChunkPersistence
 import net.corda.chunking.db.impl.persistence.StatusPublisher
 import net.corda.crypto.core.SecureHashImpl
+import net.corda.crypto.core.avro.toAvro
 import net.corda.crypto.core.parseSecureHash
-import net.corda.crypto.core.toAvro
 import net.corda.data.chunking.Chunk
 import net.corda.messaging.api.records.Record
 import net.corda.v5.base.exceptions.CordaRuntimeException

diff --git a/components/crypto/crypto-client-hsm-impl/build.gradle b/components/crypto/crypto-client-hsm-impl/build.gradle
@@ -20,6 +20,7 @@ dependencies {
     implementation project(":components:crypto:crypto-component-core-impl")
     implementation project(":libs:configuration:configuration-core")
     implementation project(":libs:crypto:crypto-impl")
+    implementation project(":libs:crypto:crypto-impl-utils")
     implementation project(":libs:messaging:messaging")
     implementation project(":libs:utilities")
 

diff --git a/...nt-hsm-impl/src/main/kotlin/net/corda/crypto/client/hsm/impl/HSMRegistrationClientImpl.kt b/...nt-hsm-impl/src/main/kotlin/net/corda/crypto/client/hsm/impl/HSMRegistrationClientImpl.kt
@@ -2,8 +2,8 @@ package net.corda.crypto.client.hsm.impl
 
 import net.corda.crypto.component.impl.retry
 import net.corda.crypto.component.impl.toClientException
-import net.corda.crypto.impl.createWireRequestContext
-import net.corda.crypto.impl.toWire
+import net.corda.crypto.impl.utils.createWireRequestContext
+import net.corda.crypto.impl.utils.toWire
 import net.corda.data.crypto.wire.CryptoNoContentValue
 import net.corda.data.crypto.wire.hsm.HSMAssociationInfo
 import net.corda.data.crypto.wire.hsm.registration.HSMRegistrationRequest

diff --git a/components/crypto/crypto-client-impl/build.gradle b/components/crypto/crypto-client-impl/build.gradle
@@ -20,7 +20,9 @@ dependencies {
     implementation project(":components:crypto:crypto-component-core-impl")
     implementation project(":libs:configuration:configuration-core")
     implementation project(':libs:crypto:cipher-suite-impl')
+    implementation project(':libs:crypto:crypto-impl-utils')
     implementation project(':libs:crypto:crypto-core')
+    implementation project(":libs:crypto:crypto-core-avro")
     implementation project(":libs:messaging:messaging")
     implementation project(':libs:platform-info')
     implementation project(":libs:utilities")

diff --git a/...to/crypto-client-impl/src/main/kotlin/net/corda/crypto/client/impl/CryptoOpsClientImpl.kt b/...to/crypto-client-impl/src/main/kotlin/net/corda/crypto/client/impl/CryptoOpsClientImpl.kt
@@ -13,8 +13,8 @@ import net.corda.crypto.core.SecureHashImpl
 import net.corda.crypto.core.ShortHash
 import net.corda.crypto.core.bytes
 import net.corda.crypto.core.publicKeyIdFromBytes
-import net.corda.crypto.impl.createWireRequestContext
-import net.corda.crypto.impl.toWire
+import net.corda.crypto.impl.utils.createWireRequestContext
+import net.corda.crypto.impl.utils.toWire
 import net.corda.data.KeyValuePairList
 import net.corda.data.crypto.SecureHashes
 import net.corda.data.crypto.ShortHashes
@@ -49,6 +49,7 @@ import java.nio.ByteBuffer
 import java.security.PublicKey
 import java.time.Duration
 import java.util.UUID
+import net.corda.crypto.core.avro.toAvro
 
 @Suppress("TooManyFunctions")
 class CryptoOpsClientImpl(
@@ -64,9 +65,6 @@ class CryptoOpsClientImpl(
             this.hashes.map {
                 SecureHashImpl(it.algorithm, it.bytes.array())
             }
-
-        private fun SecureHash.toAvro(): net.corda.data.crypto.SecureHash =
-            net.corda.data.crypto.SecureHash(this.algorithm, ByteBuffer.wrap(bytes))
     }
 
     fun getSupportedSchemes(tenantId: String, category: String): List<String> {
@@ -454,4 +452,4 @@ private fun PublicKey.fullId(keyEncodingService: KeyEncodingService, digestServi
     digestService.hash(
         keyEncodingService.encodeAsByteArray(this),
         DigestAlgorithmName.SHA2_256
-    )
+    )
diff --git a/components/crypto/crypto-service-impl/build.gradle b/components/crypto/crypto-service-impl/build.gradle
@@ -32,6 +32,7 @@ dependencies {
     implementation project(":libs:configuration:configuration-core")
     implementation project(':libs:crypto:cipher-suite-impl')
     implementation project(':libs:crypto:crypto-config-impl')
+    implementation project(":libs:crypto:crypto-impl-utils")
     implementation project(':libs:crypto:crypto-flow')
     implementation project(':libs:flows:external-event-responses')
     implementation project(":libs:messaging:messaging")

diff --git a/...to-service-impl/src/main/kotlin/net/corda/crypto/service/impl/SigningKeyInfoExtensions.kt b/...to-service-impl/src/main/kotlin/net/corda/crypto/service/impl/SigningKeyInfoExtensions.kt
@@ -0,0 +1,20 @@
+package net.corda.crypto.service.impl
+
+import java.nio.ByteBuffer
+import net.corda.crypto.cipher.suite.KeyEncodingService
+import net.corda.crypto.core.SigningKeyInfo
+import net.corda.data.crypto.wire.CryptoSigningKey
+
+fun SigningKeyInfo.toCryptoSigningKey(keyEncodingService: KeyEncodingService): CryptoSigningKey = CryptoSigningKey(
+    this.id.value,
+    this.tenantId,
+    this.category,
+    this.alias,
+    this.hsmAlias,
+    ByteBuffer.wrap(keyEncodingService.encodeAsByteArray(this.publicKey)),
+    this.schemeCodeName,
+    this.wrappingKeyAlias,
+    this.encodingVersion,
+    this.externalId,
+    this.timestamp
+)
diff --git a/...o-service-impl/src/main/kotlin/net/corda/crypto/service/impl/bus/CryptoOpsBusProcessor.kt b/...o-service-impl/src/main/kotlin/net/corda/crypto/service/impl/bus/CryptoOpsBusProcessor.kt
@@ -11,8 +11,9 @@ import net.corda.crypto.core.KeyOrderBy
 import net.corda.crypto.core.SecureHashImpl
 import net.corda.crypto.core.ShortHash
 import net.corda.crypto.impl.retrying.CryptoRetryingExecutor
-import net.corda.crypto.impl.toMap
-import net.corda.crypto.impl.toSignatureSpec
+import net.corda.crypto.impl.utils.toMap
+import net.corda.crypto.impl.utils.toSignatureSpec
+import net.corda.crypto.service.impl.toCryptoSigningKey
 import net.corda.data.crypto.SecureHashes
 import net.corda.data.crypto.ShortHashes
 import net.corda.data.crypto.wire.CryptoDerivedSharedSecret

diff --git a/...-service-impl/src/main/kotlin/net/corda/crypto/service/impl/rpc/CryptoFlowOpsProcessor.kt b/...-service-impl/src/main/kotlin/net/corda/crypto/service/impl/rpc/CryptoFlowOpsProcessor.kt
@@ -7,8 +7,8 @@ import net.corda.crypto.core.SecureHashImpl
 import net.corda.crypto.core.ShortHash
 import net.corda.crypto.core.publicKeyIdFromBytes
 import net.corda.crypto.impl.retrying.CryptoRetryingExecutor
-import net.corda.crypto.impl.toMap
-import net.corda.crypto.impl.toSignatureSpec
+import net.corda.crypto.impl.utils.toMap
+import net.corda.crypto.impl.utils.toSignatureSpec
 import net.corda.crypto.service.CryptoExceptionCategorizer
 import net.corda.crypto.service.CryptoExceptionType
 import net.corda.data.KeyValuePairList
@@ -38,6 +38,7 @@ import org.slf4j.LoggerFactory
 import java.nio.ByteBuffer
 import java.time.Duration
 import java.time.Instant
+import net.corda.crypto.service.impl.toCryptoSigningKey
 
 @Suppress("LongParameterList")
 class CryptoFlowOpsProcessor(

diff --git a/...vice-impl/src/test/kotlin/net/corda/crypto/service/impl/bus/CryptoOpsBusProcessorTests.kt b/...vice-impl/src/test/kotlin/net/corda/crypto/service/impl/bus/CryptoOpsBusProcessorTests.kt
@@ -26,7 +26,7 @@ import org.junit.jupiter.api.BeforeAll
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.TestInstance
 import java.time.Instant
-import java.util.*
+import java.util.UUID
 import java.util.concurrent.CompletableFuture
 import kotlin.test.assertEquals
 import kotlin.test.assertNotNull

diff --git a/...ice-impl/src/test/kotlin/net/corda/crypto/service/impl/rpc/CryptoFlowOpsProcessorTests.kt b/...ice-impl/src/test/kotlin/net/corda/crypto/service/impl/rpc/CryptoFlowOpsProcessorTests.kt
@@ -26,7 +26,6 @@ import net.corda.crypto.service.impl.infra.assertClose
 import net.corda.data.KeyValuePairList
 import net.corda.data.crypto.wire.CryptoResponseContext
 import net.corda.data.crypto.wire.CryptoSignatureWithKey
-import net.corda.data.crypto.wire.CryptoSigningKey
 import net.corda.data.crypto.wire.CryptoSigningKeys
 import net.corda.data.crypto.wire.ops.flow.FlowOpsRequest
 import net.corda.data.crypto.wire.ops.flow.FlowOpsResponse
@@ -54,7 +53,6 @@ import org.mockito.kotlin.doReturn
 import org.mockito.kotlin.eq
 import org.mockito.kotlin.mock
 import org.mockito.kotlin.whenever
-import java.nio.ByteBuffer
 import java.security.PublicKey
 import java.time.Instant
 import java.util.UUID
@@ -481,8 +479,5 @@ class CryptoFlowOpsProcessorTests {
         }
         on { timestamp } doAnswer { Instant.now() }
         on { publicKey } doAnswer { key0 }
-        on { toCryptoSigningKey(any()) } doAnswer { mock<CryptoSigningKey> {
-            on { publicKey } doAnswer  { ByteBuffer.wrap(keyEncodingService.encodeAsByteArray(key0)) }
-        } }
     }
-}
+}
diff --git a/...ce/src/integrationTest/kotlin/net/corda/flow/testing/tests/ExternalEventAcceptanceTest.kt b/...ce/src/integrationTest/kotlin/net/corda/flow/testing/tests/ExternalEventAcceptanceTest.kt
@@ -1,12 +1,12 @@
 package net.corda.flow.testing.tests
 
 import net.corda.data.KeyValuePairList
-import net.corda.data.flow.event.external.ExternalEventContext
 import net.corda.data.flow.event.external.ExternalEventResponseErrorType
 import net.corda.data.flow.output.FlowStates
 import net.corda.data.persistence.EntityRequest
 import net.corda.data.persistence.EntityResponse
 import net.corda.data.persistence.FindEntities
+import net.corda.flow.external.events.ExternalEventContext
 import net.corda.flow.external.events.factory.ExternalEventFactory
 import net.corda.flow.external.events.factory.ExternalEventRecord
 import net.corda.flow.fiber.FlowIORequest
@@ -15,6 +15,7 @@ import net.corda.flow.state.FlowCheckpoint
 import net.corda.flow.testing.context.ALICE_FLOW_KEY_MAPPER
 import net.corda.flow.testing.context.FlowServiceTestBase
 import net.corda.flow.testing.context.flowResumedWithError
+import net.corda.flow.utils.toAvro
 import net.corda.schema.configuration.FlowConfig
 import net.corda.v5.base.exceptions.CordaRuntimeException
 import org.junit.jupiter.api.BeforeEach
@@ -48,7 +49,7 @@ class ExternalEventAcceptanceTest : FlowServiceTestBase() {
         val ANY_INPUT = EntityRequest(
             ALICE_HOLDING_IDENTITY,
             FindEntities("entity class name", listOf(BYTE_BUFFER)),
-            ExternalEventContext(REQUEST_ID, FLOW_ID1, KeyValuePairList(emptyList()))
+            ExternalEventContext(REQUEST_ID, FLOW_ID1, emptyMap()).toAvro()
         )
         val ANY_RESPONSE = EntityResponse(listOf(BYTE_BUFFER), KeyValuePairList(emptyList()), null)
         const val STRING_INPUT = "this is an input string"

diff --git a/.../net/corda/flow/application/crypto/external/events/CreateSignatureExternalEventFactory.kt b/.../net/corda/flow/application/crypto/external/events/CreateSignatureExternalEventFactory.kt
@@ -3,10 +3,11 @@ package net.corda.flow.application.crypto.external.events
 import net.corda.crypto.core.DigitalSignatureWithKey
 import net.corda.crypto.flow.CryptoFlowOpsTransformer
 import net.corda.data.crypto.wire.ops.flow.FlowOpsResponse
-import net.corda.data.flow.event.external.ExternalEventContext
+import net.corda.flow.external.events.ExternalEventContext
 import net.corda.flow.external.events.factory.ExternalEventFactory
 import net.corda.flow.external.events.factory.ExternalEventRecord
 import net.corda.flow.state.FlowCheckpoint
+import net.corda.flow.utils.toAvro
 import net.corda.v5.base.annotations.CordaSerializable
 import net.corda.v5.crypto.SignatureSpec
 import org.osgi.service.component.annotations.Activate
@@ -33,7 +34,7 @@ class CreateSignatureExternalEventFactory @Activate constructor(
             signatureSpec = parameters.signatureSpec,
             data = parameters.bytes,
             context = emptyMap(),
-            flowExternalEventContext = flowExternalEventContext
+            flowExternalEventContext = flowExternalEventContext.toAvro()
         )
         return ExternalEventRecord(payload = flowOpsRequest)
     }

diff --git a/...lin/net/corda/flow/application/crypto/external/events/FilterMyKeysExternalEventFactory.kt b/...lin/net/corda/flow/application/crypto/external/events/FilterMyKeysExternalEventFactory.kt
@@ -2,10 +2,11 @@ package net.corda.flow.application.crypto.external.events
 
 import net.corda.crypto.flow.CryptoFlowOpsTransformer
 import net.corda.data.crypto.wire.ops.flow.FlowOpsResponse
-import net.corda.data.flow.event.external.ExternalEventContext
+import net.corda.flow.external.events.ExternalEventContext
 import net.corda.flow.external.events.factory.ExternalEventFactory
 import net.corda.flow.external.events.factory.ExternalEventRecord
 import net.corda.flow.state.FlowCheckpoint
+import net.corda.flow.utils.toAvro
 import org.osgi.service.component.annotations.Activate
 import org.osgi.service.component.annotations.Component
 import org.osgi.service.component.annotations.Reference
@@ -28,7 +29,7 @@ class FilterMyKeysExternalEventFactory @Activate constructor(
                 .createFilterMyKeys(
                     tenantId = checkpoint.holdingIdentity.shortHash.value,
                     candidateKeys = parameters,
-                    flowExternalEventContext = flowExternalEventContext
+                    flowExternalEventContext = flowExternalEventContext.toAvro()
                 )
         return ExternalEventRecord(payload = flowOpsRequest)
     }

diff --git a/...a/flow/application/persistence/external/events/AbstractPersistenceExternalEventFactory.kt b/...a/flow/application/persistence/external/events/AbstractPersistenceExternalEventFactory.kt
@@ -1,11 +1,12 @@
 package net.corda.flow.application.persistence.external.events
 
-import net.corda.data.flow.event.external.ExternalEventContext
 import net.corda.data.persistence.EntityRequest
 import net.corda.data.persistence.EntityResponse
+import net.corda.flow.external.events.ExternalEventContext
 import net.corda.flow.external.events.factory.ExternalEventFactory
 import net.corda.flow.external.events.factory.ExternalEventRecord
 import net.corda.flow.state.FlowCheckpoint
+import net.corda.flow.utils.toAvro
 import net.corda.virtualnode.toAvro
 import java.nio.ByteBuffer
 
@@ -25,12 +26,12 @@ abstract class AbstractPersistenceExternalEventFactory<PARAMETERS : Any> :
             payload = EntityRequest.newBuilder()
                 .setHoldingIdentity(checkpoint.holdingIdentity.toAvro())
                 .setRequest(createRequest(parameters))
-                .setFlowExternalEventContext(flowExternalEventContext)
+                .setFlowExternalEventContext(flowExternalEventContext.toAvro())
                 .build()
         )
     }
 
     override fun resumeWith(checkpoint: FlowCheckpoint, response: EntityResponse): List<ByteBuffer> {
         return response.results
     }
-}
+}
diff --git a/...lin/net/corda/flow/application/persistence/external/events/FindAllExternalEventFactory.kt b/...lin/net/corda/flow/application/persistence/external/events/FindAllExternalEventFactory.kt
@@ -1,13 +1,14 @@
 package net.corda.flow.application.persistence.external.events
 
-import net.corda.data.flow.event.external.ExternalEventContext
 import net.corda.data.persistence.EntityRequest
 import net.corda.data.persistence.EntityResponse
 import net.corda.data.persistence.FindAll
+import net.corda.flow.external.events.ExternalEventContext
 import net.corda.flow.external.events.factory.ExternalEventFactory
 import net.corda.flow.external.events.factory.ExternalEventRecord
 import net.corda.flow.persistence.query.OffsetResultSetExecutor
 import net.corda.flow.state.FlowCheckpoint
+import net.corda.flow.utils.toAvro
 import net.corda.utilities.toByteArrays
 import net.corda.v5.base.annotations.CordaSerializable
 import net.corda.virtualnode.toAvro
@@ -27,7 +28,7 @@ class FindAllExternalEventFactory: ExternalEventFactory<FindAllParameters, Entit
             payload = EntityRequest.newBuilder()
                 .setHoldingIdentity(checkpoint.holdingIdentity.toAvro())
                 .setRequest(FindAll(parameters.entityClass.canonicalName, parameters.offset, parameters.limit))
-                .setFlowExternalEventContext(flowExternalEventContext)
+                .setFlowExternalEventContext(flowExternalEventContext.toAvro())
                 .build()
         )
     }

diff --git a/.../net/corda/flow/application/persistence/external/events/NamedQueryExternalEventFactory.kt b/.../net/corda/flow/application/persistence/external/events/NamedQueryExternalEventFactory.kt
@@ -1,15 +1,16 @@
 package net.corda.flow.application.persistence.external.events
 
-import net.corda.data.flow.event.external.ExternalEventContext
 import net.corda.data.persistence.EntityRequest
 import net.corda.data.persistence.EntityResponse
 import net.corda.data.persistence.FindWithNamedQuery
-import net.corda.utilities.toByteArrays
-import net.corda.utilities.toByteBuffers
+import net.corda.flow.external.events.ExternalEventContext
 import net.corda.flow.external.events.factory.ExternalEventFactory
 import net.corda.flow.external.events.factory.ExternalEventRecord
 import net.corda.flow.persistence.query.OffsetResultSetExecutor
 import net.corda.flow.state.FlowCheckpoint
+import net.corda.flow.utils.toAvro
+import net.corda.utilities.toByteArrays
+import net.corda.utilities.toByteBuffers
 import net.corda.v5.base.annotations.CordaSerializable
 import net.corda.virtualnode.toAvro
 import org.osgi.service.component.annotations.Component
@@ -29,7 +30,7 @@ class NamedQueryExternalEventFactory : ExternalEventFactory<NamedQueryParameters
                 .setHoldingIdentity(checkpoint.holdingIdentity.toAvro())
                 .setRequest(FindWithNamedQuery(parameters.queryName, parameters.parameters.toByteBuffers(), parameters.offset, parameters
                     .limit, null))
-                .setFlowExternalEventContext(flowExternalEventContext)
+                .setFlowExternalEventContext(flowExternalEventContext.toAvro())
                 .build()
         )
     }