package Ġops . kafka . h arness Ċ Ċ import Ġops . kafka . Module IO Ċ import Ġops . kafka . Options Ċ import Ġops . protocols . kafka . generated . Kafka Topic Event Ċ import Ġops . protocols . kafka . generated . Kafka Topic Snapshot Ċ import Ġops . protocols . kafka . generated . Kafka User Event Ċ import Ġops . protocols . kafka . generated . Kafka User Snapshot Ċ import Ġorg . slf 4 j . Logger Ċ import Ġorg . slf 4 j . LoggerFactory Ċ import Ġorg . apache . flink . state fun . flink . h arness . Harness Ġas ĠState fun Harness Ċ Ċ object ĠH arness Ġ{ ĊĠĠĠ Ġprivate Ġval Ġlog : ĠLogger Ġ= ĠLoggerFactory . getLogger ( Harness :: class . java ) ĊĊĠĠĠ Ġ@ Jvm Static ĊĠĠĠ Ġfun Ġmain ( args : ĠArray < String >) Ġ{ ĊĠĠĠĠĠĠĠ Ġlog . info (" Starting Ġdevelopment Ġh arness ") ĊĠĠĠĠĠĠĠ Ġif Ġ( args . contains ("-- generator ")) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġgenerator Harness (). start () ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdocker Compose Harness ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠOptions . from Args ( args ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ). start () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġgenerator Harness () Ġ= ĠState fun Harness () ĊĠĠĠĠĠĠĠ Ġ. with F link Job Name (" state fun - kafka ") ĊĠĠĠĠĠĠĠ Ġ. with Configuration (" parallel ism . default ", Ġ" 1 ") ĊĠĠĠĠĠĠĠ Ġ. with Supply ing Ingress ( Module IO . K AF KA _ USER _ REQUEST _ ING RESS _ ID , ĠKafka UserRequest Generator ()) ĊĠĠĠĠĠĠĠ Ġ. with Supply ing Ingress ( Module IO . K AF KA _ TOPIC _ REQUEST _ ING RESS _ ID , ĠKafka Topic Request Generator ( 2 0 0 0 0 L )) ĊĠĠĠĠĠĠĠ Ġ. with Supply ing Ingress ( Module IO . PRO ME THE US _ METRIC _ ING RESS _ ID , ĠPrometheus Metric Generator ( 1 0 0 0 0 L )) ĊĠĠĠĠĠĠĠ Ġ. with Consum ing E gress ( ĊĠĠĠĠĠĠĠĠĠĠĠ ĠModule IO . K AF KA _ USER _ EVENT _ EG RESS _ ID , ĊĠĠĠĠĠĠĠĠĠĠĠ ĠStd out Print ing Consumer < Kafka User Event >(" USER ĠEVENT >") ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ. with Consum ing E gress ( ĊĠĠĠĠĠĠĠĠĠĠĠ ĠModule IO . K AF KA _ USER _ SNAPSHOT _ EG RESS _ ID , ĊĠĠĠĠĠĠĠĠĠĠĠ ĠStd out Print ing Consumer < Kafka User Snapshot >(" USER ĠS NAP SHOT >") ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ. with Consum ing E gress ( ĊĠĠĠĠĠĠĠĠĠĠĠ ĠModule IO . K AF KA _ TOPIC _ EVENT _ EG RESS _ ID , ĊĠĠĠĠĠĠĠĠĠĠĠ ĠStd out Print ing Consumer < Kafka Topic Event >(" TOPIC ĠEVENT >") ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ. with Consum ing E gress ( ĊĠĠĠĠĠĠĠĠĠĠĠ ĠModule IO . K AF KA _ TOPIC _ SNAPSHOT _ EG RESS _ ID , ĊĠĠĠĠĠĠĠĠĠĠĠ ĠStd out Print ing Consumer < Kafka Topic Snapshot >(" TOPIC ĠS NAP SHOT >") ĊĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠ Ġfun Ġdocker Compose Harness ( options : ĠOptions ): ĠState fun Harness Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġh arness Ġ= ĠState fun Harness () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. with F link Job Name (" state fun - kafka ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. with Configuration (" parallel ism . default ", Ġ" 1 ") ĊĊĠĠĠĠĠĠĠ Ġoptions . forEach Ġ{ Ġ( key , Ġvalue ) Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġh arness . with Global Configuration ( key , Ġvalue ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġreturn Ġh arness ĊĠĠĠ Ġ} Ċ } Ċ