['   \n  \n [\'ਮੈਂ ਫੈਸਲੇ ਦੇ ਮਾਡਲ ਲਈ ਇੱਥੇ ਆਮ ਤਰਕ ਨੂੰ ਕੱਢਣ ਲਈ ਮੌਜੂਦਾ ਕੋਡ ਨੂੰ ਦੁਬਾਰਾ ਲਿਖਣਾ ਚਾਹੁੰਦਾ ਹਾਂ, ਅਤੇ ਇਸ ਦੇ ਆਮ ਢਾਂਚੇ ਨੂੰ ਬਿਹਤਰ ਬਣਾਉਣ ਲਈ ਕਿਸੇ ਵੀ ਵਿਸ਼ੇਸ਼ ਕੋਡ ਨੂੰ ਅੱਗੇ ਵਧਾਉਣਾ ਚਾਹੁੰਦਾ ਹਾਂ।ਐਪਰਨਟਾਈਮ, ਕੀ ਤੁਸੀਂ ਇਸ ਵਿੱਚ ਮੇਰੀ ਮਦਦ ਕਰ ਸਕਦੇ ਹੋ? \\n "\\n" "\\n ਵੈਲ ਰਾਅ ਆਡੀਅੰਸ ਈਵੈਂਟ ਸਟ੍ਰੀਮਃ ਡਾਟਾਸਟ੍ਰੀਮ [ਈਵੈਂਟ ਰੈਪਰ [ਰਾਅ ਆਡੀਅੰਸ ਈਵੈਂਟ]] = ਨੂੰ ਮੂਵ ਕਰਨ ਲਈ ਕੋਡ ਦਾ ਹਿੱਸਾਕਿਨੇਸਿਸ ਸਟ੍ਰੀਮ ਸੋਰਸ \\n. ਫਲੈਟਮੈਪ (ਨਵਾਂ ਏਰਰ ਹੈਂਡਲਰ [ਰਾਅ ਆਡੀਅੰਸ ਈਵੈਂਟ] ("ਡੀਕੋਡਿੰਗ ਡਿਸੀਜ਼ਨ ਮਾਡਲ") \\n\\n ਫੈਸਲਾ ਮਾਡਲ ਸਟ੍ਰੀਮਃਡਾਟਾਸਟ੍ਰੀਮ [ਈਵੈਂਟ ਰੈਪਰ [ਜਾਂ ਤਾਂ [ਡੀਲੀਟ ਈਵੈਂਟ, ਡਿਸੀਜ਼ਨ ਮਾਡਲ]] = ਰਾਅ ਆਡੀਅੰਸ ਈਵੈਂਟ ਸਟ੍ਰੀਮ\\ਨ. ਮੈਪ (ਨਵਾਂ ਈਵੈਂਟ ਮੈਪ ਫੰਕਸ਼ਨ [ਰਾਅ ਆਡੀਅੰਸ ਈਵੈਂਟ,ਜਾਂ ਤਾਂ [ਡੀਲੀਟ ਈਵੈਂਟ, ਡੈਸੀਸ਼ਨ ਮਾਡਲ]] (\\n TopPersonaTransformer.transformTealiumEvent \\n) \\n\\n ਵੈਲ ਪ੍ਰਾਪਤ ਕਰੋਕਰੰਟ ਟਾਈਮ ਸਟੈਂਪ = () =>DateTimeFormatter.ISO_INSTANT.format (Instant.now ()) \\n ਵਾਲ ਪਰਸੋਨਾਐਲਿਮੈਂਟ ਕਨਵਰਟਰਃ ਪਰਸੋਨਾਐਲਿਮੈਂਟ ਕਨਵਰਟਰ = ਨਵਾਂਪਰਸੋਨਾਐਲਿਮੈਂਟ ਕਨਵਰਟਰ (ਗੇਟਕਰੰਟ ਟਾਈਮ ਸਟੈਂਪ) \\n\\n decisionModelStream.connect (ਹਟਾਏ ਗਏ ਲਾਕਆਈਡਸਟ੍ਰੀਮ) \\n. ਕੀਬਾਈ (_.event.fold (_.lockeId.value,_.lockeId.value), _.event.value) \\n. ਪ੍ਰੋਸੈੱਸ (ਨਵਾਂ ਕਨੈਕਟਡ ਸਟ੍ਰੀਮਫਿਲਟਰ) \\n. ਸਿੰਕਟੂ (ਡਿਸੀਸ਼ਨਿੰਗ ਡਾਇਨਾਮੋਡੀਬੀਸਿੰਕ (ਪਰਸੋਨਾਐਲੀਮੈਂਟਕਨਵਰਟਰ,config.awsEndpoint) \\n\\n env.execute () \\n} \\n "" \\n\\n ਇਸ ਤਰ੍ਹਾਂ ਦੀ ਕਿਸੇ ਚੀਜ਼ ਵੱਲ ਵਧੋ, XXX ਨੂੰ ਅਨੁਕੂਲ ਨਾਮ ਨਾਲ ਬਦਲੋਃ \\n\\n "" \\n\\ntrait XXX {\\nਟ੍ਰਾਂਸਫੋਰਮਟੇਲੀਅਮਈਵੈਂਟਃ ਰਾਅ ਆਡੀਅੰਸਈਵੈਂਟ => ਜਾਂ ਤਾਂ [ਡੀਲੀਟਈਵੈਂਟ, ਟੀ] \\ਨ, \\ਨ ਪਰਸੋਨਾਈਲਿਮੈਂਟ ਕਨਵਰਟਰ =?? \\ਨ} \\ਨ \\ਨ ਕਲਾਸ ਟਾਪਪਰਸੋਨਾਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸ{\\n\\n} \\n ਕਲਾਸ ਡੈਸੀਜ਼ਨ ਮਾਡਲ ਪ੍ਰੋਸੈਸਰ [T] (XXX: XXX) {\\n ਡਿਫ ਪ੍ਰਕਿਰਿਆ (ਰਾਅ ਆਡੀਅੰਸ ਈਵੈਂਟਸਟ੍ਰੀਮਃਡਾਟਾਸਟ੍ਰੀਮ [ਈਵੈਂਟ ਰੈਪਰ [ਰਾਅ ਆਡੀਅੰਸ ਈਵੈਂਟ], ਹਟਾਏ ਗਏ ਲਾਕਆਈਡਸਟ੍ਰੀਮਃ ਡਾਟਾਸਟ੍ਰੀਮ [ਈਵੈਂਟ ਰੈਪਰ [ਲਾਕਆਈਡੀ]) = {\\n??? \\n} \\n "" \\n\\nFYI ਦਾ ਦੂਜਾ ਹਿੱਸਾਸਿਸਟਮ \\n\\n "" \\n ਕਲਾਸ ਐਪਰਨਟਾਈਮ (ਸੰਰਚਨਾਃ ਸੰਰਚਨਾ, env: ਸਟ੍ਰੀਮਐਕਸਕਿਊਸ਼ਨ ਵਾਤਾਵਰਣ) {\\n ਡਿਫ ਸਟਾਰਟ (): ਯੂਨਿਟ = {\\n ਵੈਲ ਕਾਫਕਾਟੋਪਿਕ = "ਲੋਕ-ਈਮੇਲ" \\nਵੈਲ ਕਾਫਕਾ ਸੋਰਸ = ਕਾਫਕਾ ਸੋਰਸ ਬਿਲਡਰ [ਲਾਕਆਈਡ ਈਮੇਲ ਮੈਪਿੰਗ] (ਕਾਫਕਾਟੋਪਿਕ, config.kafkaConfig) \\n ਵੈਲ ਵਾਟਰਮਾਰਕ ਰਣਨੀਤੀ =WatermarkStrategy.noWatermarks [ਈਵੈਂਟ ਰੈਪਰ [ਏਰਰਓਆਰ [ਲਾਕਆਈਡੀਈਮੇਲਮੈਪਿੰਗ]] () \\n\\n ਵੈਲ ਕਾਫਕਾ ਸਟ੍ਰੀਮ ਸੋਰਸਃਡਾਟਾਸਟ੍ਰੀਮ [ਈਵੈਂਟ ਰੈਪਰ [ਗਲਤੀ ਜਾਂ [ਲਾਕਆਈਡਈਮੇਲਮੈਪਿੰਗ]] = \\n env.fromSource (ਕਾਫਕਾ ਸੋਰਸ, ਵਾਟਰਮਾਰਕ ਸਟ੍ਰੈਟੇਜੀ, s "$ਕਾਫਕਾਟੋਪਿਕ-ਸੋਰਸ") \\n ਵੈਲਹਟਾਏ ਗਏ ਲਾਕਆਈਡਸਟ੍ਰੀਮਃ ਡਾਟਾਸਟ੍ਰੀਮ [ਈਵੈਂਟ ਰੈਪਰ [ਲਾਕਆਈਡੀ]] = ਕਾਫਕਾ ਸਟ੍ਰੀਮ ਸੋਰਸ\\ਨ. ਫਲੈਟਮੈਪ (ਨਵਾਂ ਏਰਰਹੈਂਡਲਰ [ਲਾਕਆਈਡਈਮੇਲਮੈਪਿੰਗ] (s "ਸੋਰਸ $ਕਾਫਕਾਟੋਪਿਕਗਲਤੀਆਂ ਨੂੰ ਸੰਭਾਲੋ)) \\n. ਫਿਲਟਰ (_.event.email.isEmpty) \\n. ਮੈਪ (ਮੁੱਲ => value.map (_.lockeId). ਮੈਪ (LockeId.apply)) \\n ਵੈਲ ਕਾਇਨਸਿਸਸਟ੍ਰੀਮਸੋਰ੍ਸਃਡਾਟਾਸਟ੍ਰੀਮ [ਈਵੈਂਟ ਰੈਪਰ [ਏਰਰਓਆਰ [ਰਾਅ ਆਡੀਅੰਸ ਈਵੈਂਟ]] = \\n env.addSource (ਕਿਨੇਸਿਸ ਸੋਰਸ (config.awsEndpoint, config.kinesisStreamName)) \\n ਵੈਲਰਾਅ ਆਡੀਅੰਸ ਈਵੈਂਟ ਸਟ੍ਰੀਮਃ ਡਾ \'] ', 'ਮੈਂ ਫੈਸਲੇ ਦੇ ਮਾਡਲ ਲਈ ਇੱਥੇ ਆਮ ਤਰਕ ਨੂੰ ਕੱਢਣ ਲਈ ਮੌਜੂਦਾ ਕੋਡ ਨੂੰ ਦੁਬਾਰਾ ਲਿਖਣਾ ਚਾਹੁੰਦਾ ਹਾਂ, ਅਤੇ ਇਸ ਦੇ ਆਮ ਢਾਂਚੇ ਨੂੰ ਬਿਹਤਰ ਬਣਾਉਣ ਲਈ ਕਿਸੇ ਵੀ ਵਿਸ਼ੇਸ਼ ਕੋਡ ਨੂੰ ਅੱਗੇ ਵਧਾਉਣਾ ਚਾਹੁੰਦਾ ਹਾਂ।ਐਪਰਨਟਾਈਮ, ਕੀ ਤੁਸੀਂ ਇਸ ਵਿੱਚ ਮੇਰੀ ਮਦਦ ਕਰ ਸਕਦੇ ਹੋ? ਕੋਡ ਦਾ ਹਿੱਸਾ "ਵੈਲ ਰਾਅ ਆਡੀਅੰਸ ਈਵੰਟਸਟ੍ਰੀਮਃ ਡਾਟਾਸਟ੍ਰੀਮ [ਈਵੈਂਟ ਰੈਪਰ [ਰਾਅ ਆਡੀਅੰਸ ਈਵੰਟ]] = ਨੂੰ ਮੂਵ ਕਰਨ ਲਈ ਹੈ।ਕਿਨੇਸਿਸਸਟ੍ਰੀਮ ਸੋਰਸ. ਫਲੈਟਮੈਪ (ਨਵਾਂ ਏਰਰਹੈਂਡਲਰ [ਰਾਅ ਆਡੀਅੰਸ ਈਵੈਂਟ] ("ਡੀਕੋਡਿੰਗ ਡਿਸੀਜ਼ਨ ਮਾਡਲ")) ਵੈਲ ਫੈਸਲਾ ਮਾਡਲਸਟ੍ਰੀਮਃਡਾਟਾਸਟ੍ਰੀਮ [ਈਵੈਂਟ ਰੈਪਰ [ਜਾਂ ਤਾਂ [ਡੀਲੀਟ ਈਵੈਂਟ, ਡੈਸੀਸ਼ਨ ਮਾਡਲ]] = ਰਾਅ ਆਡੀਅੰਸ ਈਵੈਂਟ ਸਟ੍ਰੀਮ. ਮੈਪ (ਨਵਾਂ ਈਵੈਂਟ ਮੈਪ ਫੰਕਸ਼ਨ [ਰਾਅ ਆਡੀਅੰਸ ਈਵੈਂਟ, ਜਾਂ ਤਾਂ [ਡੀਲੀਟ ਈਵੈਂਟ,ਡਿਸੀਜ਼ਨ ਮਾਡਲ]] (TopPersonaTransformer.transformTealiumEvent)) ਵੈਲ ਪ੍ਰਾਪਤ ਕਰੋCurrentTimestamp = () => DateTimeFormatter.ISO_INSTANT.format (Instant.now ())ਵੈਲ ਪਰਸੋਨਾਈਲੀਮੈਂਟਕਨਵਰਟਰਃ ਪਰਸੋਨਾਈਲੀਮੈਂਟਕਨਵਰਟਰ = ਨਵਾਂ ਪਰਸੋਨਾਈਲੀਮੈਂਟਕਨਵਰਟਰ (ਗੇਟਕਰੰਟ ਟਾਈਮ ਸਟੈਂਪ)decisionModelStream.connect (ਹਟਾਏ ਗਏ ਲਾਕਆਈਡਸਟ੍ਰੀਮ). ਕੀਬਾਈ (_.event.fold (_.lockeId.value, _.lockeId.value), _.event.value). ਪ੍ਰਕਿਰਿਆ (ਨਵੀਂ)ਕਨੈਕਟਡ ਸਟ੍ਰੀਮਫਿਲਟਰ). ਸਿੰਕਟੂ (ਡਿਸੀਸ਼ਨਿੰਗ ਡਾਇਨਾਮੋਡੀਬੀਐਸਿੰਕ (ਪਰਸੋਨਾਐਲਿਮੈਂਟ ਕਨਵਰਟਰ, config.awsEndpoint)) env.execute ()} "" ਇਸ ਤਰ੍ਹਾਂ ਦੀ ਕਿਸੇ ਚੀਜ਼ ਵੱਲ ਵਧੋ,XXX ਨੂੰ ਅਨੁਕੂਲ ਨਾਮ ਨਾਲ ਬਦਲੋਃ "" ਵਿਸ਼ੇਸ਼ਤਾ XXX {ਟ੍ਰਾਂਸਫੋਰਮਟੇਲੀਅਮ ਈਵੈਂਟਃ ਰਾਅ ਆਡੀਅੰਸ ਈਵੈਂਟ => ਜਾਂ ਤਾਂ [ਡੀਲੀਟ ਈਵੈਂਟ, ਟੀ], ਪਰਸੋਨਾ ਐਲੀਮੈਂਟਕਨਵਰਟਰ =?????ਕਲਾਸ ਟਾਪਪਰਸੋਨਾਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਐਕਸਡਾਟਾਸਟ੍ਰੀਮ [ਈਵੈਂਟ ਰੈਪਰ [ਰਾਅ ਆਡੀਅੰਸ ਈਵੈਂਟ]], ਹਟਾਏ ਗਏ ਲਾਕਆਈਡਸਟ੍ਰੀਮਃ ਡਾਟਾਸਟ੍ਰੀਮ [ਈਵੈਂਟ ਰੈਪਰ [ਲਾਕਆਈਡੀ]) = {???}} "" ਐੱਫਵਾਈਆਈ ਸਿਸਟਮ ਦਾ ਦੂਜਾ ਹਿੱਸਾ ਹੈ।"" "ਕਲਾਸ ਐਪਰਨਟਾਈਮ (ਸੰਰਚਨਾਃ ਸੰਰਚਨਾ, env: ਸਟ੍ਰੀਮਐਕਸਕਿਊਸ਼ਨ ਵਾਤਾਵਰਣ) {ਡਿਫ ਸਟਾਰਟ (): ਯੂਨਿਟ = {ਵੈਲ ਕਾਫਕਾਟੋਪਿਕ =" "ਲੋਕ-ਈਮੇਲ" "ਵੈਲ ਕਾਫਕਾ ਸੋਰਸ ="ਕਾਫ਼ਕਾ ਸੋਰਸ ਬਿਲਡਰ [ਲਾਕਆਈਡਈਮੇਲਮੈਪਿੰਗ] (ਕਾਫ਼ਕਾਟੋਪਿਕ, config.kafkaConfig) ਵੈਲ ਵਾਟਰਮਾਰਕਸਟ੍ਰੈਟੇਜੀ =WatermarkStrategy.noWatermarks [ਈਵੈਂਟ ਰੈਪਰ [ਏਰਰਓਆਰ [ਲਾਕਆਈਡੀਈਮੇਲਮੈਪਿੰਗ]] () ਵੱਲੋਂ ਕਾਫਕਾ ਸਟ੍ਰੀਮ ਸੋਰਸਃਡਾਟਾਸਟ੍ਰੀਮ [ਈਵੈਂਟ ਰੈਪਰ [ਗਲਤੀ ਜਾਂ [ਲਾਕਆਈਡਈਮੇਲਮੈਪਿੰਗ]] = env.fromSource (ਕਾਫਕਾ ਸੋਰਸ, ਵਾਟਰਮਾਰਕ ਸਟ੍ਰੈਟੇਜੀ, s "$ਕਾਫਕਾਟੋਪਿਕ-ਸੋਰਸ") ਵਾਲਹਟਾਏ ਗਏ ਲਾਕਆਈਡਸਟ੍ਰੀਮਃ ਡਾਟਾਸਟ੍ਰੀਮ [ਈਵੈਂਟ ਰੈਪਰ [ਲਾਕਆਈਡੀ]] = ਕਾਫਕਾ ਸਟ੍ਰੀਮ ਸੋਰਸ. ਫਲੈਟਮੈਪ (ਨਵਾਂ ਏਰਰਹੈਂਡਲਰ [ਲਾਕਆਈਡਈਮੇਲਮੈਪਿੰਗ] (s "ਸੋਰਸ $ਕਾਫਕਾਟੋਪਿਕ ਹੈਂਡਲ")ਗਲਤੀਆਂ "). ਫਿਲਟਰ (_.event.email.isEmpty). ਮੈਪ (ਮੁੱਲ => value.map (_.lockeId). ਮੈਪ (LockeId.apply)) ਵੈਲ ਕਿਨੇਸਿਸਸਟ੍ਰੀਮਸੋਰਸਃਡਾਟਾਸਟ੍ਰੀਮ [ਈਵੈਂਟ ਰੈਪਰ [ਗਲਤੀ ਜਾਂ [ਰਾਅ ਆਡੀਅੰਸ ਈਵੈਂਟ]] = env.addSource (ਕਿਨੇਸਿਸ ਸੋਰਸ (config.awsEndpoint, config.kinesisStreamName)) ਵੈਲਰਾਅ ਆਡੀਅੰਸ ਈਵੈਂਟ ਸਟ੍ਰੀਮਃ ਡਾ \n ']