21 Creating Kafka Topics for ECE Integration

You create Apache Kafka topics so Oracle Communications Elastic Charging Engine (ECE) can support sending Notifications with HTTP Gateway, Diameter Gateway, RADIUS Gateway and BRM Gateway.

Topics in this document:

Setting Up ECE and Gateways for Apache Kafka Using Topic-Based Routing

You can use the enhanced Kafka integration in Oracle Communications Billing and Revenue Management (BRM) for improved real-time data processing, particularly in managing high data volumes. A dedicated message processor ensures efficient and reliable system functionality.

Note:

This is the recommended configuration for Kafka integration.

The message processor handles data in a controlled and sequential manner. Its checkpointing mechanism enables reliable recovery and uninterrupted data processing during system interruptions. If a gateway goes out of service, the system automatically redirects its partitions to another gateway without manual intervention. When the original gateway returns, traffic is automatically rerouted back.

To set up the Apache Kafka architecture using topic-based routing, follow these steps:

Note:

For new multisite deployments, install and configure all sites at the same time, since there is no traffic in the system yet.

  1. Set the kafkaCustomPartitionAssignorEnabled ECE configuration MBean to true. See "Enabling the Topic-Based Kafka Architecture".

  2. Wait until all ECS instances or pods have been restarted and any rebalancing has completed.

  3. Perform a rolling restart of the Diameter Gateway, RADIUS Gateway, HTTP Gateway, EM Gateway, and BRM Gateway so that the consumers are restarted, and they subscribe to the new Kafka topics.

  4. On multisite deployments, repeat steps 2 through 4 until all sites have been configured to use the Kafka topic-based routing feature.

  5. On multisite deployments, restart the ReplicatedFederatedCache service between the sites in the Coherence cluster and verify that the ECE Configuration AppConfiguration MBeans in all the sites are in sync.

Enabling the Topic-Based Kafka Architecture

By default, ECE uses the Apache Kafka partition-based architecture to communicate with network clients, but you can configure ECE to use the topic-based architecture instead.

To enable the topic-based Kafka architecture without restarting ECE:
  1. Do one of the following based on your system type:

  2. Expand the ECE Configuration node.

  3. Expand charging.kafkaConfigurations.

  4. Expand Attributes.

  5. Set the kafkaCustomPartitionAssignorEnabled attribute to true.

Alternatively, you can use the charging-settings.xml file or your override-values.yaml file for oc-cn-ece-helm-chart to set the kafkaCustomPartitionAssignorEnabled attribute. If you make the change by updating the file, you must perform a rolling restart of your ECE application instances or pods for the change to take effect.

Generating Kafka Topics with Topic-Based Routing

You use the create_all_kafka_topics.sh script to create the required Kafka topics. When running the script, use the Kafka broker details for all sites.

To generate Kafka topics with topic-based routing:

  1. Ensure you have installed the Apache Kafka command-line tools, such as kafka-topics.sh, and that their location is in the PATH environment variable or the KAFKA_HOME and KAFKA_BIN_DIR environment variables.

  2. Do one of the following based on your system type:

    • Cloud native systems: Create an override-values.yaml file for oc-cn-ece-helm-chart and configure the keys under kafkaConfigurationList to meet your system requirements.

    • On-premises systems: In your charging-settings.xml file, configure the kafkaConfigurationList entries to meet your system requirements.

  3. In your configuration file, configure the Kafka brokers to allow automatic topic creation. Set the default number of partitions for new topics to either 20 or the product of the number of Kafka broker instances and the number of gateways that will consume notifications for a topic, whichever is greater.

    For example, if you have 3 Kafka brokers and 8 Diameter Gateways, set the default to 24 (3 x 8) partitions. This configuration ensures that there are more partitions than consumers, allowing each consumer to have its own partition and distributing partitions evenly across Kafka brokers for optimal resource utilization.

    Note:

    For multisite deployments, the configuration file should have the details for all sites. This ensures that the create_all_kafka_topics.sh script creates all necessary topics for all Kafka brokers across all sites, supporting automatic high availability and disaster recovery.

  4. Run the create_all_kafka_topics.sh script: 

    create_all_kafka_topics.sh overrideValuesFile --bootstrap-server kafkaHost:port --partitions numPartitions [--ssl enabled | disabled] [--truststore-location trustStorePath ] [--replication-factor replicationNum]

    where:

    • overrideValuesFile specifies the name and path to the override-values.yaml file you created in step 2.

    • kafkaHost and port specify the host and port that the Kafka client will connect to in a bootstrap Kafka cluster the first time it starts.

    • numPartitions specifies the number of partitions to create in the topic.

    • trustStorePath specifies the path to the TrustStore file. This option is required only if SSL is enabled.

    • replicationNum specifies how many copies of each topic’s partition are stored across different brokers in the Kafka cluster. The default is 1, which disables the replication.

Enter the trustStorePassword when prompted by the script, where trustStorePassword is the password for the TrustStore. This behavior applies when SSL is enabled.

Migrating Existing ECE Systems Gateways to Apache Kafka Using Topic-Based Routing

To migrate from the existing integration architecture to the Apache Kafka topic-based routing:

  1. For multisite deployments, stop the federation between the site about to be migrated and the other sites.

  2. Set the kafkaCustomPartitionAssignorEnabled ECE configuration MBean to true. See "Enabling the Topic-Based Kafka Architecture".

    This updates the charging-settings.xml file in the cache and updates the ECE charging-settings-namespace ConfigMap.

  3. During the ECE postinstallation process, run the create_all_kafka_topics.sh script to create the required Kafka topics. See "Generating Kafka Topics with Topic-Based Routing".

    Wait until all ECS instances or pods have been restarted and any rebalancing has completed.

  4. Drain the remaining notifications from the Kafka brokers at the site being migrated. Using the existing site's gateways, any remaining notifications in the old EceNotifications topics at the Kafka broker will be consumed.

    Run this step until all notifications on the old topics are consumed. Wait until the old topics have been drained of all remaining notifications so that the consumers are restarted and subscribe to the new Kafka topics.

  5. Perform a rolling restart of the Diameter Gateway, RADIUS Gateway, HTTP Gateway, EM Gateway, and BRM Gateway so that the consumers are restarted, and they subscribe to the new Kafka topics.

    Note:

    Ensure that BRM is pointed to another site’s EM Gateways prior to bringing down the EM Gateway at the site being migrated.

  6. On multisite deployments, repeat 2 through 6 for all the sites until they have been migrated to using the new Kafka redesign feature.

  7. Restart the configuration federation of the ReplicatedFederatedCache service between the sites in the Coherence cluster and verify that the ECE Configuration AppConfiguration MBeans in all the sites are in sync.

    Note:

    For ECS pods, run the helm upgrade with the job to update app configuration enabled.

Setting Up ECE and Gateways for Apache Kafka Using Partition-Based Routing

Note:

This configuration is recommended for backward compatibility only.

To set up ECE with your Kafka topics:

  1. When you install ECE:

    • Specify to use Apache Kafka topics and enter the details for your ECE notification topics, Suspense topic, ECE failure topic, and ECE overage topic.

    • (HTTP Gateway Only) Enable Network Repository Function (NRF) registration in one of your HTTP Gateway servers.

    • (BRM Gateway Only) Specify the details for connecting to the BRM Gateway.

    For more information, see "Installing Elastic Charging Engine" in ECE Installation Guide.

  2. During the ECE postinstallation process:

    • Run the kafka_post_install.sh script to create your ECE notification topics, Suspense topic, ECE failure topic, and ECE overage topic. See "Creating Kafka Topics for ECE" in ECE Installation Guide.

    • Run the post_Install.sh script and choose to create only the Acknowledgment queue. See "Creating WebLogic JMS Queues for BRM" in ECE Installation Guide.

    Note:

    It is strongly recommended that you use the kafka_post_install.sh script to create all ECE notification topics in advance instead of relying on topic auto-creation. This ensures that ECE notification consumers will be able to subscribe to topics even before the first notification message is produced.

    All ECE consumer groups are prefixed with Ece. Your consumers should never use a consumer group that starts with this prefix to avoid collisions.

  3. Configure Kafka partition assignments for gateways. See "Configuring Kafka Partition Assignments for Gateways".

  4. (Optional) Configure ECE to send information about failed usage requests to the ECE failure topic. See "Recording Failed ECE Usage Requests".

  5. (Optional) Configure ECE to generate CDRs for any prepaid usage overage and send them to the ECE overage topic. See "Understanding and Configuring ECE Prepaid Usage Overage".

Configuring Kafka Partition Assignments for Gateways

Configure Kafka partition assignments in ECE AppConfiguration; do not configure them in kafka_post_install.sh. For cloud native deployments, set kafkaPartition and kafkaPartitionForInternalExternal in your override-values.yaml file for oc-cn-ece-helm-chart. For on-premises deployments, set the attributes in the ECE Configuration MBeans by using a JMX editor such as JConsole.

To configure Kafka partition assignments for gateways:

  1. For each Diameter Gateway and BRM Gateway configuration, set kafkaPartition to one dedicated Kafka partition ID.

  2. For each HTTP Gateway configuration, set kafkaPartition to a comma-separated list of dedicated Kafka partition IDs.

    Configure at least one partition ID for each HTTP Gateway replica that might run.

    For example, if httpgw1 is configured with 1,2,3, httpgw1-0 consumes notifications from partition 1, httpgw1-1 consumes notifications from partition 2, and httpgw1-2 consumes notifications from partition 3.

  3. Set kafkaPartitionForInternalExternal to the dedicated partition ID for notifications configured for internal and external publishing.

  4. Ensure that every partition ID specified by kafkaPartition or kafkaPartitionForInternalExternal is unique across the ECE deployment.

    Also ensure that the ECE notification topic contains enough partitions for all of the partition IDs that you assign. See "Connecting ECE and Gateways to Kafka Topics."

  5. Do one of the following:

    • For cloud native deployments, after you update the Kafka partition application configuration values, run the chargingConfigurationReloader job to reload the application configuration. See "Reloading ECE Application Configuration ChangesReloading ECE Application Configuration Changes" in BRM Cloud Native System Administrator's Guide.

    • For on-premises deployments, update the kafkaPartition and kafkaPartitionForInternalExternal attributes by using a JMX editor such as JConsole.

Determining the Number of Partitions Required

Use this procedure only for Kafka partition-based routing. Determine the minimum number of partitions in the ECE notification topic before you create the topic and assign partition IDs to gateways.

To determine the minimum number of partitions:
  1. Identify the maximum number of dedicated gateway notification consumers that might run in the deployment:

    • Diameter Gateway instances

    • HTTP Gateway replicas across all HTTP Gateway configurations

    • BRM Gateway instances

    • One consumer for notifications configured by kafkaPartitionForInternalExternal

  2. If you assign consecutive partition IDs starting with 0, calculate the minimum number of partitions by using this formula:

    [(Max Diameter Gateway instances) + (Max HTTP Gateway replicas) + (Max BRM Gateway instances) + (1 for kafkaPartitionForInternalExternal)]

    For example, if you have 4 Diameter Gateway instances, 3 HTTP Gateway replicas, and 1 BRM Gateway instance, create at least [(4 + 3 + 1 + 1) = 9] partitions. This example assumes that the partition IDs are 0 through 8.

  3. Compare the calculated minimum with the highest partition ID that you plan to assign. The number of partitions must be greater than the highest partition ID configured in kafkaPartition or kafkaPartitionForInternalExternal.

  4. Configure the partitions value to the larger of the calculated minimum and the highest partition ID plus 1.

Note:

If the ECE External Notification topic is enabled, configure its partition count separately by using externalPartitions. You can set partitions higher than the minimum to allow future gateway partition IDs without changing the Kafka broker.

Connecting ECE and Gateways to Kafka Topics

You connect ECE to the following Kafka topics so that ECE can publish notifications, failed usage requests, and CDRs with usage overage information to them.

  • ECE notification topics: Stores notifications from ECE.

  • Suspense topic: Stores failed notifications from ECE.

  • ECE failure topic: (HTTP Gateway and Diameter Gateway only) Stores details about failed ECE usage requests, such as the user ID and request payload.

  • ECE overage topic: (HTTP Gateway and Diameter Gateway only) Stores overage records, which contain details about usage overage amounts for prepaid customers.

You can also connect the HTTP Gateway, Diameter Gateway, and RADIUS Gateway to these Kafka topics so they can retrieve notifications, failed usage requests, and CDRs with usage overage information from them.

Note:

If you are using an ECE system with a topic-based Kafka architecture, ECE splits the ECE notifications topic into multiple granular topics based on factors such as cluster, gateway, notification type, and client.

ECE creates each notification topic using this naming convention:

Notify.clusterName.consumerGroup[.notificationName][.networkClientId][.eceGatewayId]

where:

  • clusterName is the string that represents the site name from your charging-settings.xml file. The allowable values are BRM1, BRM2, ECE1, ECE2, and so on.

  • consumerGroup is the name of the consumer group. The possible values are DiamNetwork, RadNetwork, ChfNetwork, Brm, Ext, Suspense, and Overage.

  • notificationName is the notification event type. This is included for DiamNetwork, RadNetwork, ChfNetwork, Brm, and Ext consumer groups.

  • networkClientId is used to differentiate the CHF Network (for the HTTP Gateway) from the Diameter Client (for the Diameter Gateway). This is included for DiamNetwork and ChfNetwork consumer groups.

  • eceGatewayId is the value taken from the Diameter Gateway ID. This is included for DiamNetwork consumer groups.

To connect ECE and gateways to your Kafka topics:

  1. Access the ECE configuration MBeans in a JMX editor, such as JConsole. See "Accessing ECE Configuration MBeans".

  2. Expand the ECE Configuration node.

  3. Expand charging.kafkaConfigurations.

  4. Expand Attributes.

  5. Specify the Kafka configuration values for the attributes in Table 21-1.

    Table 21-1 kafkaConfiguration Properties

    Property Name Description

    name

    The name of your ECE cluster.

    hostname

    The host name and port number of the machine in which Apache Kafka is installed.

    If it contains multiple Kafka brokers, create a comma-separated list.

    topicName

    The name of the Kafka topic where ECE will publish notifications.

    Note: This is for partition-based routing only.

    suspenseTopicName

    The name of the Kafka topic where failed notifications are published.

    Note: This is for partition-based routing only.

    failureTopicName

    The name of the Kafka topic where ECE will publish details about failed usage requests.

    Note: This is for partition-based routing only.

    overageTopicName

    The name of the Kafka topic where ECE will publish overage records with information about your prepaid customer's usage overage during online sessions.

    Note: This is for partition-based routing only.

    partitions

    The number of partitions in the ECE notification topic for Kafka partition-based routing. Configure enough partitions for all gateway notification consumers that use dedicated partitions, plus the partition configured by kafkaPartitionForInternalExternal.

    For the calculation, examples, and validation rule, see "Determining the Number of Partitions Required."

    Note: Note: This is for partition-based routing only.

    externalPartitions

    The total number of Kafka partitions in the ECE External Notification topic.

    Note: This is for partition-based routing only.

    externalTopicEnabled

    The parameter that specifies, whether ECE External Notification topic is created and in use. Set to true to enable.

    externalTopicName

    The name of the ECE External Notification topic.

    Note: This is for partition-based routing only.

    failurePartitions

    The number of Kafka partitions in your ECE failure topic.

    Note: This is for partition-based routing only.

    kafkaCustomPartitionAssignorEnabled

    The parameter to enable topic-based routing for Kafka notifications which uses a custom partition assignor for automated support for high availability and disaster recovery.

    If this parameter is set to false, the partition-based routing will be used.

    kafkaProducerReconnectionInterval

    The amount of time, in milliseconds, the ECE Notification Publisher waits before attempting to reconnect to a Kafka topic.

    kafkaProducerReconnectionMax

    The maximum amount of time, in milliseconds, the ECE Notification Publisher waits before attempting to reconnect to a broker that has repeatedly failed to connect.

    The kafkaProducerReconnectionInterval will increase exponentially for each consecutive connection failure, up to this maximum.

    kafkaHTTPReconnectionInterval

    The amount of time, in milliseconds, HTTP Gateway waits before attempting to reconnect to the Kafka topic.

    kafkaHTTPReconnectionMax

    The maximum amount of time, in milliseconds, HTTP Gateway waits before attempting to reconnect to a broker that has repeatedly failed to connect.

    The kafkaHTTPReconnectionInterval will increase exponentially for each consecutive connection failure, up to this maximum.

    kafkaDGWReconnectionInterval

    The amount of time, in milliseconds, Diameter Gateway waits before attempting to reconnect to the Kafka topic.

    kafkaDGWReconnectionMax

    The maximum amount of time, in milliseconds, Diameter Gateway waits before attempting to reconnect to a broker that has repeatedly failed to connect.

    The kafkaDGWReconnectionInterval will increase exponentially for each consecutive connection failure, up to this maximum.

    kafkaBRMReconnectionInterval

    The amount of time, in milliseconds, BRM Gateway waits before attempting to reconnect to the Kafka topic.

    kafkaBRMReconnectionMax

    The maximum amount of time, in milliseconds, BRM Gateway waits before attempting to reconnect to a broker that has repeatedly failed to connect.

    The kafkaBRMReconnectionInterval will increase exponentially for each consecutive connection failure, up to this maximum.

    kafkaFederationIssuesTopicName

    The Kafka topic name for federation-related issues.

    kafkaFederationIssuesReconnectionInterval

    The reconnection interval, in milliseconds, of the Kafka producer to the federation issues topic.

    kafkaFederationIssuesReconnectionMax

    The maximum reconnection time, in milliseconds, of the Kafka producer to the federation issues topic

    kafkaRGWTopicName

    The Kafka topic that is used for Radius Gateway notifications.

    kafkaRGWReconnectionIntervalMillis

    The amount of time, in milliseconds, RADIUS Gateway waits before attempting to reconnect to the Kafka topic.

    kafkaRGWReconnectionMaxMillis

    The maximum amount of time, in milliseconds, RADIUS Gateway waits before attempting to reconnect to a broker that has repeatedly failed to connect.

    The kafkaRGWReconnectionIntervalMillis will increase exponentially for each consecutive connection failure up to this maximum.

    walletLocation

    The path of the wallet where certificates are stored when connecting to the Kafka brokers using SSL.

Recording Failed ECE Usage Requests

ECE may occasionally fail to process usage requests. For example, a data usage request could fail because a customer has insufficient funds. You can configure ECE to publish details about failed usage requests, such as the user ID and request payload, to the ECE failure topic in your Kafka server. Later on, you can reprocess the usage requests or view the failure details for analysis and reporting.

To enable the recording of failed ECE usage requests:

  1. Access the ECE configuration MBeans in a JMX editor, such as JConsole.

  2. Expand the ECE Configuration node.

  3. Expand charging.kafkaConfigurations.

  4. Expand Attributes.

  5. Set the persistFailedRequestsToKafkaTopic property to true.

Note:

You can also examine the gateway log files to determine why a usage request failed. See "Troubleshooting Failed Usage Requests" in BRM System Administrator's Guide for more information.

Understanding and Configuring ECE Prepaid Usage Overage

ECE can generate overage records when prepaid usage exceeds the amount that was granted, reserved, or available for the session. These overage records help account for usage that ECE cannot fully charge during the active session and help prevent revenue leakage.

When the network reports that the number of Used Service Units (USUs) is greater than the number of Granted Service Units (GSUs), ECE treats the excess usage as usage beyond the granted or reserved amount. In this flow, ECE applies built-in credit-protection behavior:

  1. ECE charges the customer only up to the available allowance.

  2. ECE generates an overage record for the excess usage.

  3. ECE sends the overage record to the ECE Overage Kafka topic.

For example, assume that a customer has a prepaid balance of 100 minutes, but uses 130 minutes during a session. If the network reports 130 minutes of usage after only 100 minutes were granted, ECE charges the customer for 100 minutes, creates an overage record for the remaining 30 minutes of usage, and writes the overage record to the ECE Overage Kafka topic.

Note:

If your system does not contain a Kafka server, you can configure ECE to publish overage usage amounts to a log file instead. See "Logging Prepaid Usage Overage".

You can also configure ECE to create overage records during post-rating validation by using the checkReservationOverImpact attribute. When this attribute is enabled and the rated balance impact exceeds the customer’s available balance, ECE applies the balance impact only up to the available balance and creates an overage record for the remaining amount.

Use checkReservationOverImpact when you want ECE to capture overage during post-rating validation.

Note:

The built-in credit-protection behavior is not controlled by the checkReservationOverImpact attribute. Therefore, even when checkReservationOverImpact is set to false, ECE can still generate overage records when the network reports USUs are greater than GSUs.

To configure ECE to perform post-rating validation for prepaid usage overage:

  1. Ensure that you have created an ECE Overage Kafka topic and connected ECE to your Kafka server. See "Connecting ECE and Gateways to Kafka Topics".

  2. Enable ECE to perform post-rating validation and capture overage when the rated balance impact exceeds the available balance:

    1. Access the ECE configuration MBeans in a JMX editor, such as JConsole. See "Accessing ECE Configuration MBeans".

    2. Expand the ECE Configuration node.

    3. Expand charging.server.

    4. Expand Attributes.

    5. Set the checkReservationOverImpact attribute to true. The default is false.

You can also customize ECE to include any existing unrated quantity in a CDR or notification using the ECE SDK post-charging extension (postChargingExtension). Use the OverageDetailsImpl class to include the unrated quantity. The extension can then capture the unrated quantity by using the getOverageDetailsMap() method.