CORE-17768 Topology changes - Use new topics

applications/examples/sandbox-app/src/main/kotlin/net/corda/example/vnode/CordaVNode.kt

@@ -19,7 +19,7 @@ import net.corda.libs.packaging.core.CpkMetadata
 import net.corda.messaging.api.records.Record
 import net.corda.osgi.api.Application
 import net.corda.osgi.api.Shutdown
-import net.corda.schema.Schemas.Flow.FLOW_EVENT_TOPIC
+import net.corda.schema.Schemas.Flow.FLOW_START
 import net.corda.schema.configuration.ConfigKeys.FLOW_CONFIG
 import net.corda.schema.configuration.FlowConfig.PROCESSING_FLOW_CLEANUP_TIME
 import net.corda.schema.configuration.FlowConfig.PROCESSING_MAX_FLOW_SLEEP_DURATION
@@ -172,11 +172,11 @@ class CordaVNode @Activate constructor(
 
             val rpcStartFlow = createRPCStartFlow(clientId, vnodeInfo.toAvro())
             val flowId = generateRandomId()
-            val record = Record(FLOW_EVENT_TOPIC, flowId, FlowEvent(flowId, rpcStartFlow))
+            val record = Record(FLOW_START, flowId, FlowEvent(flowId, rpcStartFlow))
             flowEventProcessorFactory.create(mapOf(FLOW_CONFIG to smartConfig)).apply {
                 val result = onNext(null, record)
                 result.responseEvents.singleOrNull { evt ->
-                    evt.topic == FLOW_EVENT_TOPIC
+                    evt.topic == FLOW_START
                 }?.also { evt ->
                     @Suppress("unchecked_cast")
                     onNext(result.updatedState, evt as Record<String, FlowEvent>)

charts/corda-lib/templates/_worker.tpl

@@ -84,7 +84,7 @@ metadata:
 spec:
   type: ClusterIP
   selector:
-    app: {{ $workerName }}
+    app.kubernetes.io/component: {{ include "corda.workerComponent" $worker }}
   ports:
       - protocol: TCP
         port: {{ include "corda.workerServicePort" . }}

components/crypto/crypto-service-impl/src/test/kotlin/net/corda/crypto/service/impl/bus/CryptoFlowOpsBusProcessorTests.kt

@@ -257,7 +257,7 @@ import kotlin.test.assertTrue
                  )
              ).thenReturn(
                  Record(
-                     Schemas.Flow.FLOW_EVENT_TOPIC,
+                     Schemas.Flow.FLOW_SESSION,
                      flowExternalEventContexts.get(it).flowId,
                      FlowEvent()
                  )
@@ -269,7 +269,7 @@ import kotlin.test.assertTrue
                  )
              ).thenReturn(
                  Record(
-                     Schemas.Flow.FLOW_EVENT_TOPIC,
+                     Schemas.Flow.FLOW_SESSION,
                      flowExternalEventContexts.get(it).flowId,
                      FlowEvent()
                  )
@@ -281,7 +281,7 @@ import kotlin.test.assertTrue
                  )
              ).thenReturn(
                  Record(
-                     Schemas.Flow.FLOW_EVENT_TOPIC,
+                     Schemas.Flow.FLOW_SESSION,
                      flowExternalEventContexts.get(it).flowId,
                      FlowEvent()
                  )

components/flow/flow-mapper-impl/src/main/kotlin/net/corda/flow/mapper/impl/FlowMapperEventExecutorFactoryImpl.kt

@@ -17,7 +17,7 @@ import net.corda.flow.mapper.impl.executor.SessionEventExecutor
 import net.corda.flow.mapper.impl.executor.SessionInitProcessor
 import net.corda.flow.mapper.impl.executor.StartFlowExecutor
 import net.corda.libs.configuration.SmartConfig
-import net.corda.schema.Schemas.Flow.FLOW_EVENT_TOPIC
+import net.corda.schema.Schemas.Flow.FLOW_START
 import org.osgi.service.component.annotations.Activate
 import org.osgi.service.component.annotations.Component
 import org.osgi.service.component.annotations.Reference
@@ -66,7 +66,7 @@ class FlowMapperEventExecutorFactoryImpl @Activate constructor(
                 }
             }
 
-            is StartFlow -> StartFlowExecutor(eventKey, FLOW_EVENT_TOPIC, flowMapperEventPayload, state)
+            is StartFlow -> StartFlowExecutor(eventKey, FLOW_START, flowMapperEventPayload, state)
             is ExecuteCleanup -> ExecuteCleanupEventExecutor(eventKey)
             is ScheduleCleanup -> ScheduleCleanupEventExecutor(eventKey, flowMapperEventPayload, state)
 

components/flow/flow-mapper-impl/src/main/kotlin/net/corda/flow/mapper/impl/RecordFactoryImpl.kt

@@ -97,10 +97,10 @@ class RecordFactoryImpl @Activate constructor(
 
     private fun getSessionEventOutputTopic(sessionEvent: SessionEvent): String {
         return when (sessionEvent.messageDirection) {
-            MessageDirection.INBOUND -> Schemas.Flow.FLOW_EVENT_TOPIC
+            MessageDirection.INBOUND -> Schemas.Flow.FLOW_SESSION
             MessageDirection.OUTBOUND -> {
                 if (isLocalCluster(sessionEvent)) {
-                    Schemas.Flow.FLOW_MAPPER_EVENT_TOPIC
+                    Schemas.Flow.FLOW_MAPPER_SESSION_IN
                 } else {
                     Schemas.P2P.P2P_OUT_TOPIC
                 }
@@ -120,8 +120,8 @@ class RecordFactoryImpl @Activate constructor(
     ) : Record<*, *> {
         val outputTopic = getSessionEventOutputTopic(sourceEvent)
         val (newDirection, sessionId) = when (outputTopic) {
-            Schemas.Flow.FLOW_MAPPER_EVENT_TOPIC -> Pair(MessageDirection.INBOUND, toggleSessionId(sourceEvent.sessionId))
-            Schemas.Flow.FLOW_EVENT_TOPIC -> Pair(MessageDirection.INBOUND, sourceEvent.sessionId)
+            Schemas.Flow.FLOW_MAPPER_SESSION_IN -> Pair(MessageDirection.INBOUND, toggleSessionId(sourceEvent.sessionId))
+            Schemas.Flow.FLOW_SESSION -> Pair(MessageDirection.INBOUND, sourceEvent.sessionId)
             else -> Pair(MessageDirection.OUTBOUND, sourceEvent.sessionId)
         }
         val sequenceNumber = if (newPayload is SessionError) null else sourceEvent.sequenceNum
@@ -136,14 +136,14 @@ class RecordFactoryImpl @Activate constructor(
             sourceEvent.contextSessionProperties
         )
         return when (outputTopic) {
-            Schemas.Flow.FLOW_EVENT_TOPIC -> {
+            Schemas.Flow.FLOW_SESSION -> {
                 if (flowId == null) {
                     throw IllegalArgumentException("Flow ID is required to forward an event back to the flow event" +
                             "topic, but it was not provided.")
                 }
                 Record(outputTopic, flowId, FlowEvent(flowId, sessionEvent))
             }
-            Schemas.Flow.FLOW_MAPPER_EVENT_TOPIC -> {
+            Schemas.Flow.FLOW_MAPPER_SESSION_IN -> {
                 Record(outputTopic, sessionEvent.sessionId, FlowMapperEvent(sessionEvent))
             }
             Schemas.P2P.P2P_OUT_TOPIC -> {

components/flow/flow-mapper-impl/src/test/kotlin/net/corda/flow/mapper/impl/RecordFactoryImplTest.kt

@@ -27,7 +27,6 @@ import org.mockito.kotlin.anyOrNull
 import org.mockito.kotlin.mock
 import org.mockito.kotlin.verify
 import org.mockito.kotlin.whenever
-import java.lang.IllegalArgumentException
 import java.nio.ByteBuffer
 import java.time.Instant
 
@@ -98,7 +97,7 @@ internal class RecordFactoryImplTest {
             "my-flow-id"
         )
         assertThat(record).isNotNull
-        assertThat(record.topic).isEqualTo(Schemas.Flow.FLOW_MAPPER_EVENT_TOPIC)
+        assertThat(record.topic).isEqualTo(Schemas.Flow.FLOW_MAPPER_SESSION_IN)
         assertThat(record.value!!::class).isEqualTo(FlowMapperEvent::class)
         verify(locallyHostedIdentitiesServiceSameCluster).isHostedLocally(bobId.toCorda())
         val sessionOutput = (record.value as FlowMapperEvent).payload as SessionEvent
@@ -165,7 +164,7 @@ internal class RecordFactoryImplTest {
             flowConfig,
             FLOW_ID
         )
-        assertThat(record.topic).isEqualTo(Schemas.Flow.FLOW_EVENT_TOPIC)
+        assertThat(record.topic).isEqualTo(Schemas.Flow.FLOW_SESSION)
         assertThat(record.key).isEqualTo(FLOW_ID)
         assertThat(record.value!!::class.java).isEqualTo(FlowEvent::class.java)
         val sessionOutput = (record.value as FlowEvent).payload as SessionEvent
@@ -194,7 +193,7 @@ internal class RecordFactoryImplTest {
             FLOW_ID
         )
         assertThat(record).isNotNull
-        assertThat(record.topic).isEqualTo(Schemas.Flow.FLOW_MAPPER_EVENT_TOPIC)
+        assertThat(record.topic).isEqualTo(Schemas.Flow.FLOW_MAPPER_SESSION_IN)
         assertThat(record.value!!::class).isEqualTo(FlowMapperEvent::class)
         val sessionOutput = (record.value as FlowMapperEvent).payload as SessionEvent
         assertThat(sessionOutput.sessionId).isEqualTo("$SESSION_ID-INITIATED")
@@ -249,7 +248,7 @@ internal class RecordFactoryImplTest {
             flowConfig,
             FLOW_ID
         )
-        assertThat(record.topic).isEqualTo(Schemas.Flow.FLOW_EVENT_TOPIC)
+        assertThat(record.topic).isEqualTo(Schemas.Flow.FLOW_SESSION)
         assertThat(record.key).isEqualTo(FLOW_ID)
         assertThat(record.value!!::class.java).isEqualTo(FlowEvent::class.java)
         val sessionOutput = (record.value as FlowEvent).payload as SessionEvent
@@ -316,7 +315,7 @@ internal class RecordFactoryImplTest {
             timestamp,
             flowConfig,
         )
-        assertThat(record.topic).isEqualTo(Schemas.Flow.FLOW_MAPPER_EVENT_TOPIC)
+        assertThat(record.topic).isEqualTo(Schemas.Flow.FLOW_MAPPER_SESSION_IN)
         assertThat(record.key).isEqualTo("$SESSION_ID-INITIATED")
         assertThat(record.value!!::class).isEqualTo(FlowMapperEvent::class)
         val sessionOutput = (record.value as FlowMapperEvent).payload as SessionEvent

components/flow/flow-mapper-impl/src/test/kotlin/net/corda/flow/mapper/impl/executor/SessionInitProcessorTest.kt

@@ -29,7 +29,7 @@ class SessionInitProcessorTest {
             flowId: String
         ): Record<*, *> {
             return if (sourceEvent.messageDirection == MessageDirection.INBOUND) {
-                Record(Schemas.Flow.FLOW_EVENT_TOPIC, flowId, FlowEvent(flowId, sourceEvent))
+                Record(Schemas.Flow.FLOW_SESSION, flowId, FlowEvent(flowId, sourceEvent))
             } else {
                 Record(Schemas.P2P.P2P_OUT_TOPIC, "sessionId", "")
             }
@@ -79,7 +79,7 @@ class SessionInitProcessorTest {
 
         Assertions.assertThat(outboundEvents.size).isEqualTo(1)
         val outboundEvent = outboundEvents.first()
-        Assertions.assertThat(outboundEvent.topic).isEqualTo(Schemas.Flow.FLOW_EVENT_TOPIC)
+        Assertions.assertThat(outboundEvent.topic).isEqualTo(Schemas.Flow.FLOW_SESSION)
         Assertions.assertThat(outboundEvent.key::class).isEqualTo(String::class)
         Assertions.assertThat(outboundEvent.value!!::class).isEqualTo(FlowEvent::class)
         Assertions.assertThat(payload.sessionId).isEqualTo("sessionId-INITIATED")