We're using Kafka Streams in the SWATCH Billable Usage service to group usages within an hour that then will be sent to the SWATCH Producer AWS/Azure/Red Hat Marketplace services.

However, we're seeing lots of inconveniences of using Kafka Streams for this. The most relevant ones are:

• Missing billing usages to be grouped: SWATCH-2908
• Kafka and database transaction issues with Quarkus
• The Kafka Streams Quarkus extension is not officially supported by Quarkus

In order to improve the SWATCH Billable Usage service's resiliency and observability we need to replace the Kafka Streams implementation and use the supported Kafka Quarkus extension.

Context

At the moment, the hourly aggregation of billable usages works as described in the following diagram:

1. The SWATCH Tally service sends tally summary messages to the “platform.rhsm-subscriptions.tally” topic.
2. The SWATCH Billable Usage service consumes the tally summary messages and for each one:
   1. Map the tally summary message into a billable usage message (logic here)
   2. Populate the contract coverage if usage is contract enabled (logic here)
   3. Map the billable usage message into the billable_usage_remittance entity (check schema in the diagram, logic here)
   4. Persist the new entity into the database
   5. Send the billable usage message into the “platform.rhsm-subscriptions.billable-usage” topic
3. The SWATCH Billable Usage service consumes the billable usage message and for each one:
   1. Kafka Streams keeps the messages and aggregates them per hourly basis
   2. When the time window is closed, it sends a billable usage aggregate message into the “platform.rhsm-subscriptions.billable-usage-hourly-aggregate” topic.

Changes

The workflow would look like as:

1. Cron Job is executed hourly to invoke a BillableUsage API
   1. Gather all the records "billable_usage_remittance" and status is pending and group these records based on the existing "BillableUsageAggregateKey" class and hourly. Filter out the records older than a concrete amount of time defined in configuration.
   2. Update the above records with status “In Progress” and updated_at to now().
   3. Send aggregations to the “platform.rhsm-subscriptions.billable-usage-hourly-aggregate” topic using the group-id to either “swatch-producer-aws” or “swatch-producer-azure”.
      1. If the above transaction fails, update the records to "Pending" status
      2. If the above transaction works, update the records to “Sent” status.

Links

• Presentation: https://drive.google.com/file/d/1vWLkksjJVnCOosLhRNsJDkRk7T2scgmv/view?usp=drive_web
• Document: https://docs.google.com/document/d/1qhSrQNmO1MnNRwq3Ms8fKG-zcKivZDbikGnOpCO_rZU/edit?usp=sharing
• Diagram: https://drive.google.com/file/d/1myDaC74bzA8DO6QX9IF_qUdkc6pL0qQv/view?usp=sharing