Adding a Kafka producer for sending events to the Event Bus

.github/CODEOWNERS

@@ -608,7 +608,7 @@ app/uploaders/form1010cg @department-of-veterans-affairs/vfs-10-10 @department-o
 app/uploaders/form1010cg/poa_uploader.rb @department-of-veterans-affairs/vfs-10-10 @department-of-veterans-affairs/va-api-engineers @department-of-veterans-affairs/backend-review-group
 app/uploaders/form_upload @department-of-veterans-affairs/platform-va-product-forms @department-of-veterans-affairs/va-api-engineers @department-of-veterans-affairs/backend-review-group
 app/uploaders/hca_attachment_uploader.rb @department-of-veterans-affairs/vfs-authenticated-experience-backend @department-of-veterans-affairs/va-api-engineers @department-of-veterans-affairs/backend-review-group
-app/uploaders/lighthouse_document_uploader_base.rb @department-of-veterans-affairs/benefits-management-tools-be @department-of-veterans-affairs/va-api-engineers @department-of-veterans-affairs/backend-review-group @department-of-veterans-affairs/benefits-admin 
+app/uploaders/lighthouse_document_uploader_base.rb @department-of-veterans-affairs/benefits-management-tools-be @department-of-veterans-affairs/va-api-engineers @department-of-veterans-affairs/backend-review-group @department-of-veterans-affairs/benefits-admin
 app/uploaders/lighthouse_document_uploader.rb @department-of-veterans-affairs/benefits-management-tools-be @department-of-veterans-affairs/va-api-engineers @department-of-veterans-affairs/backend-review-group @department-of-veterans-affairs/benefits-admin
 app/uploaders/preneed_attachment_uploader.rb @department-of-veterans-affairs/mbs-core-team @department-of-veterans-affairs/va-api-engineers @department-of-veterans-affairs/backend-review-group
 app/uploaders/set_aws_config.rb @department-of-veterans-affairs/va-api-engineers @department-of-veterans-affairs/backend-review-group
@@ -661,6 +661,7 @@ app/sidekiq/hca @department-of-veterans-affairs/vfs-10-10 @department-of-veteran
 app/sidekiq/identity @department-of-veterans-affairs/octo-identity
 app/sidekiq/income_limits @department-of-veterans-affairs/vfs-public-websites-frontend @department-of-veterans-affairs/va-api-engineers @department-of-veterans-affairs/backend-review-group
 app/sidekiq/in_progress_form_cleaner.rb @department-of-veterans-affairs/vfs-authenticated-experience-backend @department-of-veterans-affairs/va-api-engineers @department-of-veterans-affairs/backend-review-group
+app/sidekiq/kafka @department-of-veterans-affairs/ves-submission-traceability @department-of-veterans-affairs/backend-review-group
 app/sidekiq/kms_key_rotation @department-of-veterans-affairs/va-api-engineers @department-of-veterans-affairs/backend-review-group
 app/sidekiq/lighthouse @department-of-veterans-affairs/backend-review-group
 app/sidekiq/lighthouse/submit_career_counseling_job.rb @department-of-veterans-affairs/my-education-benefits @department-of-veterans-affairs/va-api-engineers @department-of-veterans-affairs/backend-review-group
@@ -952,6 +953,7 @@ lib/ihub @department-of-veterans-affairs/va-api-engineers @department-of-veteran
 lib/income_and_assets @department-of-veterans-affairs/pension-and-burials @department-of-veterans-affairs/backend-review-group
 lib/json_marshal @department-of-veterans-affairs/vsa-healthcare-health-quest-1-backend @department-of-veterans-affairs/va-api-engineers @department-of-veterans-affairs/backend-review-group
 lib/json_schema @department-of-veterans-affairs/va-api-engineers @department-of-veterans-affairs/backend-review-group
+lib/kafka @department-of-veterans-affairs/ves-submission-traceability @department-of-veterans-affairs/backend-review-group
 lib/lgy @department-of-veterans-affairs/benefits-non-disability @department-of-veterans-affairs/va-api-engineers @department-of-veterans-affairs/backend-review-group
 lib/lgy/configuration.rb @department-of-veterans-affairs/benefits-non-disability @department-of-veterans-affairs/va-api-engineers @department-of-veterans-affairs/backend-review-group
 lib/lgy/service.rb @department-of-veterans-affairs/benefits-non-disability @department-of-veterans-affairs/va-api-engineers @department-of-veterans-affairs/backend-review-group
@@ -1315,6 +1317,7 @@ spec/factories/persistent_attachments.rb @department-of-veterans-affairs/platfor
 spec/factories/saved_claim.rb @department-of-veterans-affairs/pension-and-burials @department-of-veterans-affairs/va-api-engineers @department-of-veterans-affairs/backend-review-group
 spec/fixtures/ask @department-of-veterans-affairs/va-api-engineers @department-of-veterans-affairs/backend-review-group
 spec/fixtures/ask/minimal.json @department-of-veterans-affairs/va-api-engineers @department-of-veterans-affairs/backend-review-group
+spec/fixtures/avro_schemas @department-of-veterans-affairs/backend-review-group @department-of-veterans-affairs/ves-submission-traceability
 spec/fixtures/bgs @department-of-veterans-affairs/va-api-engineers @department-of-veterans-affairs/backend-review-group
 spec/fixtures/carma @department-of-veterans-affairs/vfs-10-10 @department-of-veterans-affairs/va-api-engineers @department-of-veterans-affairs/backend-review-group
 spec/fixtures/carma/10-10CG_f6056cff-d4cb-4058-8fb0-42296e12698f.pdf @department-of-veterans-affairs/vfs-10-10 @department-of-veterans-affairs/va-api-engineers @department-of-veterans-affairs/backend-review-group
@@ -1371,6 +1374,7 @@ spec/fixtures/supplemental_claims @department-of-veterans-affairs/benefits-decis
 spec/fixtures/va_profile @department-of-veterans-affairs/vfs-authenticated-experience-backend @department-of-veterans-affairs/va-api-engineers @department-of-veterans-affairs/backend-review-group
 spec/fixtures/vba_documents @department-of-veterans-affairs/lighthouse-banana-peels @department-of-veterans-affairs/va-api-engineers @department-of-veterans-affairs/backend-review-group
 spec/fixtures/vbms @department-of-veterans-affairs/benefits-dependents-management @department-of-veterans-affairs/va-api-engineers @department-of-veterans-affairs/backend-review-group
+spec/lib/kafka @department-of-veterans-affairs/ves-submission-traceability @department-of-veterans-affairs/backend-review-group
 spec/lib/vye @department-of-veterans-affairs/backend-review-group @department-of-veterans-affairs/govcio-vfep-codereviewers
 spec/sidekiq/account_login_statistics_job_spec.rb @department-of-veterans-affairs/octo-identity
 spec/sidekiq/benefits_intake_status_job_spec.rb @department-of-veterans-affairs/platform-va-product-forms @department-of-veterans-affairs/Disability-Experience @department-of-veterans-affairs/va-api-engineers @department-of-veterans-affairs/backend-review-group

Gemfile

@@ -45,6 +45,8 @@ gem 'aasm'
 gem 'activerecord-import'
 gem 'activerecord-postgis-adapter'
 gem 'addressable'
+gem 'avro'
+gem 'aws-msk-iam-sasl-signer', '~> 0.1.1'
 gem 'aws-sdk-kms'
 gem 'aws-sdk-s3', '~> 1'
 gem 'aws-sdk-sns', '~> 1'
@@ -163,6 +165,7 @@ gem 'utf8-cleaner'
 gem 'vets_json_schema', git: 'https://github.com/department-of-veterans-affairs/vets-json-schema', branch: 'master'
 gem 'virtus'
 gem 'warden-github'
+gem 'waterdrop'
 gem 'will_paginate'
 gem 'with_advisory_lock'
 
@@ -193,6 +196,7 @@ group :test do
   gem 'shoulda-matchers'
   gem 'simplecov', require: false
   gem 'super_diff'
+  # gem 'testcontainers-compose'
   gem 'vcr'
   gem 'webrick'
 end

Gemfile.lock

@@ -138,7 +138,7 @@ GEM
       gserver
       sidekiq (>= 7.3.7, < 8)
       sidekiq-pro (>= 7.3.4, < 8)
-    sidekiq-pro (7.3.5)
+    sidekiq-pro (7.3.6)
       sidekiq (>= 7.3.7, < 8)
 
 GEM
@@ -235,16 +235,24 @@ GEM
       fiddle
     ast (2.4.2)
     attr_extras (7.1.0)
+    avro (1.12.0)
+      multi_json (~> 1.0)
     awesome_print (1.9.2)
     aws-eventstream (1.3.0)
-    aws-partitions (1.1048.0)
-    aws-sdk-core (3.218.1)
+    aws-msk-iam-sasl-signer (0.1.1)
+      aws-sdk-kafka
+      thor
+    aws-partitions (1.1039.0)
+    aws-sdk-core (3.216.0)
       aws-eventstream (~> 1, >= 1.3.0)
       aws-partitions (~> 1, >= 1.992.0)
       aws-sigv4 (~> 1.9)
       base64
       jmespath (~> 1, >= 1.6.1)
-    aws-sdk-kms (1.98.0)
+    aws-sdk-kafka (1.89.0)
+      aws-sdk-core (~> 3, >= 3.216.0)
+      aws-sigv4 (~> 1.5)
+    aws-sdk-kms (1.97.0)
       aws-sdk-core (~> 3, >= 3.216.0)
       aws-sigv4 (~> 1.5)
     aws-sdk-s3 (1.180.0)
@@ -619,6 +627,13 @@ GEM
     jwe (0.4.0)
     jwt (2.10.1)
       base64
+    karafka-core (2.4.8)
+      karafka-rdkafka (>= 0.17.6, < 0.19.0)
+      logger (>= 1.6.0)
+    karafka-rdkafka (0.18.1)
+      ffi (~> 1.15)
+      mini_portile2 (~> 2.6)
+      rake (> 12)
     kms_encrypted (1.6.0)
       activesupport (>= 6.1)
     kramdown (2.4.0)
@@ -1108,6 +1123,10 @@ GEM
       addressable
       faraday (>= 1.9, < 3)
       nokogiri (>= 1.13.9)
+    waterdrop (2.8.1)
+      karafka-core (>= 2.4.3, < 3.0.0)
+      karafka-rdkafka (>= 0.17.5)
+      zeitwerk (~> 2.3)
     web-console (4.2.1)
       actionview (>= 6.0.0)
       activemodel (>= 6.0.0)
@@ -1158,8 +1177,10 @@ DEPENDENCIES
   appeals_api!
   apps_api!
   ask_va_api!
+  avro
   avs!
   awesome_print
+  aws-msk-iam-sasl-signer (~> 0.1.1)
   aws-sdk-kms
   aws-sdk-s3 (~> 1)
   aws-sdk-sns (~> 1)
@@ -1349,6 +1370,7 @@ DEPENDENCIES
   virtus
   vye!
   warden-github
+  waterdrop
   web-console
   webmock
   webrick

app/sidekiq/kafka/example_job.rb

@@ -0,0 +1,15 @@
+# frozen_string_literal: true
+
+require 'kafka/avro_producer'
+module Kafka
+  class ExampleJob
+    include Sidekiq::Job
+    # Errors that might occur during the job execution are usually not retryable,
+    # though we might want to experiment with this in practice
+    sidekiq_options retry: false
+
+    def perform(topic, payload)
+      Kafka::AvroProducer.new.produce(topic, payload)
+    end
+  end
+end

config/features.yml

@@ -2043,6 +2043,9 @@ features:
   virtual_agent_enable_datadog_logging:
     actor_type: user
     description: If enabled, allows for the use of Datadog logging for the chatbot
+  kafka_producer:
+    actor_type: cookie_id
+    description: Enables the Kafka producer for the VA.gov platform
   show_about_yellow_ribbon_program:
     actor_type: user
     description: If enabled, show additional info about the yellow ribbon program

config/initializers/sidekiq.rb

@@ -49,6 +49,10 @@
     config.death_handlers << lambda do |job, ex|
       Rails.logger.error "#{job['class']} #{job['jid']} died with error #{ex.message}."
     end
+
+    config.on(:shutdown) do
+      Kafka::ProducerManager.instance.producer&.close
+    end
   end
 
   Sidekiq.configure_client do |config|

config/puma.rb

@@ -12,3 +12,7 @@
   SemanticLogger.reopen
   ActiveRecord::Base.establish_connection
 end
+
+on_worker_shutdown do
+  Kafka::ProducerManager.instance.producer&.close
+end

config/settings.yml

@@ -70,8 +70,8 @@ session_cookie:
   secure: false
 
 sign_in:
-  jwt_encode_key: 'spec/fixtures/sign_in/privatekey.pem'
-  jwt_old_encode_key: 'spec/fixtures/sign_in/privatekey_old.pem'
+  jwt_encode_key: "spec/fixtures/sign_in/privatekey.pem"
+  jwt_old_encode_key: "spec/fixtures/sign_in/privatekey_old.pem"
   cookies_secure: false
   info_cookie_domain: localhost
   auto_uplevel: true
@@ -81,7 +81,7 @@ sign_in:
   vaweb_client_id: vaweb
   vamobile_client_id: vamobile
   arp_client_id: arp
-  user_info_clients: ['okta_test']
+  user_info_clients: ["okta_test"]
   sts_client:
     base_url: http://localhost:3000
     key_path: spec/fixtures/sign_in/sts_client.pem
@@ -105,8 +105,8 @@ clamav:
   mock: false
   # host & port here are only used in development here:
   # config/initializers/clamav.rb
-  host: 'clamav'
-  port: '3310'
+  host: "clamav"
+  port: "3310"
 
 database_url: postgis:///vets-api
 test_database_url: postgis:///vets-api-test
@@ -644,8 +644,8 @@ ask_va_api:
 claims_api:
   pdf_generator_526:
     url: ~
-    path: '/form-526ez-pdf-generator/v1/forms/'
-    content_type: 'application/vnd.api+json'
+    path: "/form-526ez-pdf-generator/v1/forms/"
+    content_type: "application/vnd.api+json"
   report_enabled: false
   audit_enabled: false
   disability_claims_mock_override: false
@@ -1322,7 +1322,6 @@ vanotify:
     ivc_forms: ivc_forms_token
     benefits_disability: benefits_disability_token
 
-
 # Settings for connecting AFS Veteran Services
 # For locahost we can use the existing certs as long as we don't call out
 dgi:
@@ -1453,8 +1452,8 @@ chip:
   mock: true
   mobile_app:
     tenant_id: 6f1c8b41-9c77-469d-852d-269c51a7d380
-    username: 'testuser'
-    password: '12345'
+    username: "testuser"
+    password: "12345"
 
 check_in:
   vaos:
@@ -1677,8 +1676,8 @@ avs:
   api_jwt: abcd1234abcd1234abcd1234abcd1234abcd1234
 
 brd:
-  api_key: 'fake_key'
-  base_name: 'https://something.fake.va.gov'
+  api_key: "fake_key"
+  base_name: "https://something.fake.va.gov"
 
 nod_vanotify_status_callback:
   bearer_token: bearer_token_secret
@@ -1729,8 +1728,8 @@ vye:
     bucket: ~
 
 schema_contract:
-  appointments_index: 'modules/vaos/app/schemas/appointments_index.json'
-  test_index: 'spec/fixtures/schema_contract/test_schema.json'
+  appointments_index: "modules/vaos/app/schemas/appointments_index.json"
+  test_index: "spec/fixtures/schema_contract/test_schema.json"
 
 form_mock_ae_design_patterns:
   prefill: true
@@ -1770,3 +1769,9 @@ banners:
   drupal_username: banners_api
   drupal_password: test
   drupal_url: https://test.cms.va.gov/
+
+kafka_producer:
+  broker_urls: <%= ENV['KAFKA_PRODUCER__BROKER_URLS'] %>
+  aws_region: "us-gov-west-1"
+  aws_role_arn: <%= ENV['KAFKA_PRODUCER__AWS_ROLE_ARN'] %>
+  schema_registry_url: <%= ENV['KAFKA_PRODUCER__SCHEMA_REGISTRY_URL'] %>

config/settings/development.yml

@@ -24,3 +24,8 @@ web_origin_regex: \Ahttps?:\/\/.*\z
 modules_appeals_api:
   documentation:
     notice_of_disagreements_v1: true
+
+kafka_producer:
+  broker_urls: ["localhost:19092"] # for local Kafka cluster in Docker
+  aws_role_arn: "arn:aws:iam::123456789012:role/role-name" # this is an example
+  schema_registry_url: "http://localhost:8081" # for local Schema Registry in Docker

config/settings/test.yml

@@ -1,4 +1,3 @@
-
 hostname: www.example.com # possible fallback for unsafe request.host
 virtual_hosts: ["www.example.com", "localhost", "127.0.0.1", "example.org"] # Safe host names
 vsp_environment: test
@@ -27,20 +26,20 @@ saml_ssoe:
 clamav:
   mock: true
   host: 0.0.0.0
-  port: '33100'
+  port: "33100"
 
 sign_in:
-  jwt_encode_key: 'spec/fixtures/sign_in/privatekey.pem'
+  jwt_encode_key: "spec/fixtures/sign_in/privatekey.pem"
 
 edu:
   prefill: true
   sftp:
-    relative_path: 'spool_files'
+    relative_path: "spool_files"
 
 pension_burial:
   prefill: true
   sftp:
-    relative_path: 'VETSGOV_PENSION'
+    relative_path: "VETSGOV_PENSION"
 
 bpds:
   jwt_secret: "FAKE_SECRET"
@@ -255,13 +254,13 @@ lighthouse:
         rsa_key: path/to/key
       use_mocks: false
   benefits_education:
-      host: https://sandbox-api.va.gov
-      use_mocks: true
-      access_token:
-        client_id: ~
-        rsa_key: ~
+    host: https://sandbox-api.va.gov
+    use_mocks: true
+    access_token:
+      client_id: ~
+      rsa_key: ~
   benefits_intake:
-      api_key: fake_api_key
+    api_key: fake_api_key
 
 vaos:
   eps:
@@ -396,9 +395,9 @@ claims_api:
   evss_container:
     client_key: fakekeyvaluehere
   pdf_generator_526:
-    url: 'https://fake_pdf_url.com'
-    path: '/form-526ez-pdf-generator/v1/forms/'
-    content_type: 'application/vnd.api+json'
+    url: "https://fake_pdf_url.com"
+    path: "/form-526ez-pdf-generator/v1/forms/"
+    content_type: "application/vnd.api+json"
   benefits_documents:
     host: https://staging-api.va.gov
     auth:
@@ -432,20 +431,20 @@ nod_vanotify_status_callback:
 
 travel_pay:
   veis:
-    client_id: 'client_id'
-    client_secret: 'client_secret'
-    resource: 'resource_id'
-    tenant_id: 'tenant_id'
-    auth_url: 'https://auth.veis.gov'
+    client_id: "client_id"
+    client_secret: "client_secret"
+    resource: "resource_id"
+    tenant_id: "tenant_id"
+    auth_url: "https://auth.veis.gov"
   mock: false
-  subscription_key: 'api_key'
-  base_url: 'https://btsss.gov'
-  client_number: '12345'
-  service_name: 'BTSSS-API'
+  subscription_key: "api_key"
+  base_url: "https://btsss.gov"
+  client_number: "12345"
+  service_name: "BTSSS-API"
 
 va_notify:
   status_callback:
-    bearer_token: 'va_notify_bearer_token'
+    bearer_token: "va_notify_bearer_token"
 
 accredited_representative_portal:
   allow_list:
@@ -454,3 +453,8 @@ accredited_representative_portal:
       base_uri: https://not.real.com
       repo: faked/fakeness
       path: not_real.csv
+
+kafka_producer:
+  broker_urls: ["localhost:19092"] # for local Kafka cluster in Docker
+  aws_role_arn: "arn:aws:iam::123456789012:role/role-name" # this is an example
+  schema_registry_url: "http://localhost:8081" # for local Schema Registry in Docker