7 Processing Rated Events for Loading Using Kafka

Learn how to configure and use Event Stream Processor (ESP) to consume rated events from Kafka, process them, and store them in the Oracle Communications Billing and Revenue Management (BRM) database.

Topics in this chapter:

About the Event Stream Processor

The Event Stream Processor (ESP) consumes rated events from Kafka, processes them as needed, and stores them as zipped JSON in the BRM database. Rated Event Manager (RE Manager) loads the events and updates balances, items, and journals as applicable. ESP processing depends on the event source:

  • For internal events: ECE Rated Event Formatter (REF) publishes events in JSONFILE format. ESP processes them using the configured internal event handler and batching settings. See Table 5-9 and Table 5-13.

  • For ECE Composable Services events: ESP consumes rated events from the configured Kafka topic and passes them to the handler configured for that topic. ESP maps and enriches the events as needed. See Table 5-12, Table 5-13, and Table 5-16.

  • For external events: ESP consumes prerated events from the configured Kafka topic, maps or converts them as needed, and uses enrichment modules to supply missing BRM data. See Table 5-12, Table 5-13, Table 5-16, and Table 5-20.

Note:

ECE REF events published in ZIPFILE format follow the RE Manager Stream Processor path, which consumes zipped CDRs directly from Kafka.

You can run the following scripts to start and stop the ESP application.

  • To start the ESP, run the start_event_stream_processor script located in the BRM_home/bin directory.

  • To stop the ESP, run the stop_event_stream_processor script located in the BRM_home/bin directory.

Processing Rated Events with ESP

ESP processes rated events as follows:

  1. The event source publishes rated events to a Kafka topic. See "Event Streaming Mode" for information about the supported event sources.

  2. ESP's Kafka consumer retrieves the events and delivers them in batches to ESP. See "Configuring Kafka Consumer, Topics, and Event Handlers" for more information.

  3. ESP routes the events to the handler configured for the Kafka topic. If the topic has no specific handler mapping, ESP uses the configured default handler. See "Configuring Kafka Consumer, Topics, and Event Handlers" for more information.

  4. The handler maps the events to the BRM event structure when the event source requires mapping. See "About the Mapper Framework" for more information.

  5. ESP applies the configured enrichment modules to supply any required BRM information that the events do not contain. See "Configuring Event Enrichment" for more information.

  6. At commit time, ESP checks for duplicate events if duplicate-event checking is configured and supported for the event source. See "Preventing Duplicate Event Processing" for more information.

  7. ESP stores the accepted events as zipped JSON and records their Kafka offset information in the BRM database as part of the commit.

  8. The RE Manager ZIP processor reads the zipped JSON from the BRM database and loads the events. See "RE Manager ZIP Processor Configuration" for more information.

  9. RE Manager updates balances, items, and journals as applicable.

  10. If an event fails during processing or loading, see "Managing Failed Rated Events" for information about identifying and recovering the event.

Configuring Kafka Consumer, Topics, and Event Handlers

The Kafka consumer retrieves rated events from Kafka topics and passes them to ESP for processing. ESP routes the events from each topic to its configured event handler.

Note:

If you do not configure a matching handler for a topic, ESP uses the default handler.

To configure the Kafka consumer, topics, and event handlers in the Infranet.properties file:

  1. Open the BRM_home/apps/event_stream_processor/Infranet.properties file in a text editor.

  2. Edit the file to configure the following entries:

    • Kafka consumer: Configure the consumer properties listed in Table 5-8 and the availability and batching properties listed in Table 5-9.

    • Topics: Configure each Kafka topic and its consumer count by using the entries in Table 5-12.

    • Event handlers: Configure the handler for each topic by using the entries in Table 5-13. Use the same type and number suffix as the corresponding topic entry. Configure the default handler for topics without a matching handler entry.

    Note:

    For ECE Composable Services TMF635 v5 events, configure the topic-specific deserializer described in Table 5-12 and use the EventHandlerImpl class specified for ECE Composable Services events in Table 5-13.

  3. Save and close the file.

Kafka consumer properties in the ESP Infranet.properties file use Apache Kafka consumer configuration names with the kafka.config. prefix. For information about consumer properties not listed in Table 5-8, see https://kafka.apache.org/43/getting-started/introduction/ for your Kafka release. Add the property to the Infranet.properties file with the kafka.config. prefix.

Configuring Kafka SSL and OAuth2

ESP uses Kafka SSL properties to secure its connection to the Kafka broker with TLS. If the broker requires SASL authentication, configure the mechanism required by the broker. For more information about Kafka SSL/TLS configuration, see https://kafka.apache.org/40/security/encryption-and-authentication-using-ssl/, and for SASL mechanisms, including OAUTHBEARER, see https://kafka.apache.org/40/security/authentication-using-sasl/.

To configure Kafka SSL and OAuth2 for ESP:

  1. Open the BRM_home/apps/event_stream_processor/Infranet.properties file in a text editor.

  2. Configure the entries required by your Kafka broker:

    • SSL and TLS: Configure a truststore so ESP can validate the broker certificate when it establishes the TLS connection. If the broker requires client certificate authentication, also configure a client keystore. See Table 5-21.

    • SASL: Configure the security protocol and authentication mechanism required by the broker. For SASL authentication over TLS, set the security protocol to SASL_SSL. See Table 5-21.

    • OAuth2: If the broker uses the OAUTHBEARER mechanism, configure the token endpoint and login callback handler. See Table 5-21.

      Note:

      If you configured OAuth2, add the token endpoint URL to the Kafka OAuth allowed URLs. Also add a certificate endpoint URL if the Kafka client retrieves certificates from that endpoint:

      1. For a cloud native system: Add -Dorg.apache.kafka.sasl.oauthbearer.allowed.urls to event_stream_processor.configEnv.jvmOpts in the applicable values file. If jvmOpts already contains other JVM options, append this property to the existing value.

    • Credentials: Configure the JAAS entry for the selected SASL mechanism. Store passwords and client secrets in the BRM wallet and use wallet entry references in the Infranet.properties file. See Table 5-21.

  3. Save and close the file.

  4. Start or restart ESP.

For information about configuring OAuth2 for the Kafka DM, see "Configuring OAuth2 for the Kafka DM" in BRM Developer's Guide.

About the Mapper Framework

ESP uses the Mapper Framework to transform an incoming rated event into the BRM table and field structure required for loading. ESP obtains the path to the mapper’s application.yaml configuration file from the mapper.configuration.path entry in Infranet.properties. The application.yaml file defines routing rules that evaluate fields in the incoming event and select the appropriate mapping specification file. The selected specification defines how to extract, transform, validate, and map event values to the corresponding BRM tables and fields.

The mapping path depends on the event source:

  • For internal events: ESP processes ECE REF JSONFILE events with the installed internal event builder. These events do not require a mapping specification file.

  • For ECE Composable Services events: When the event format requires mapping, ESP uses the configured TMF scenario and mapping specification file. The Charging Gateway Function (CGF) composable service publishes these events to Kafka as CloudEvents containing TMF635 v5 ProductUsage JSON. The records contain raw BRM POIDs for account, service, product, balance group, and item references, and serialized BRM invoice data. ESP uses the supplied references and invoice data when mapping the records.

  • For external events: ESP uses the routing rules in application.yaml to identify the event scenario and select the applicable mapping specification file. If the incoming event is not in JSON format and you have configured a converter for it, ESP first invokes the corresponding format converter to convert the event to JSON. ESP then applies the selected mapping specification to transform the normalized event into the required BRM table and field structure.

Specifying the Location of the Mapper YAML File

To configure the mapper for each event source that requires mapping:

  1. Open the BRM_home/apps/event_stream_processor/Infranet.properties file in a text editor.

  2. Set mapper.configuration.path to the location of the mapper’s application.yaml file. See Table 5-16.

  3. Save and close the file.

Configuring Mapper Scenarios in application.yaml

ESP reads the application.yaml file to select a mapping specification file for an incoming event. In this file, you specify where ESP finds the mapping specification files and how it selects one for an event. You can also configure a custom format-to-JSON converter for non-JSON input.

To configure the application.yaml file:

  1. Open the application.yaml file at the path specified by mapper.configuration.path in the ESP Infranet.properties file.

  2. Configure the entries that apply to the incoming event:

    • baseMappingDir: Specifies the directory that contains the mapping specification files.

    • mapperSpecificationKeys: Specifies routing rules. Each rule identifies a field in the incoming event and maps one or more field values to a mapping specification. Specify the mapping specification name without the .yaml extension.

      For ECE Composable Services TMF635 v5 rated events, use the usageType field to select the mapping specification. ESP provides the mapping specification files listed in Table 7-1

      Table 7-1 ECE Composable Services TMF635 v5 Mapping Specification Files

      Event usageType value Mapping Specification File

      GSM

      /event/delayed/session/telco/gsm

      tmf635v5-produsage-gsm.yaml

      GPRS

      /event/delayed/session/telco/gprs

      tmf635v5-produsage-gprs.yaml

      The following example routes GSM and GPRS events to their corresponding mapping specification files:

      baseMappingDir: <CCS_mapping_directory>
mapperSpecificationKeys:
  usageType:
    "/event/delayed/session/telco/gsm": tmf635v5-produsage-gsm
    "/event/delayed/session/telco/gprs": tmf635v5-produsage-gprs

      where CCS_mapping_directory is the directory that contains the mapping specification files.

    • jsonConverters (Optional): Specifies a custom format-to-JSON converter for a non-JSON input format. Configure the converter class, its supported record types, and any converter-specific settings.

    The following example routes an event with the serviceType value /service/telco/gsm/telephony to the external-event-simple-flat-usage.yaml mapping specification. It also shows a custom CSV-to-JSON converter configuration: 

    baseMappingDir: /path/to/mapping/specifications
jsonConverters:
  partnerCsv:
    supportedRecordTypes:
      - text/csv
      - application/csv
    converterClass: com.example.mapper.PartnerCsvConverter
    configuration:
      metadataPath: /path/to/csv-metadata.yaml
mapperSpecificationKeys:
  serviceType:
    "/service/telco/gsm/telephony": external-event-simple-flat-usage

    where:

    • baseMappingDir specifies the directory containing the mapping specification files.

    • jsonConverters specifies an optional custom adapter for incoming formats that require conversion to JSON.

    • partnerCsv specifies a user-defined, unique name for this adapter configuration. It is only a label and it does not need to match the Java class name.

    • supportedRecordTypes lists the incoming event content types handled by the adapter.

    • converterClass specifies the fully qualified name of the custom adapter class.

    • configuration contains adapter-specific settings. Mapper Framework passes these settings to the adapter, whose implementation defines their meaning.

    • metadataPath is an example adapter-specific setting that identifies a CSV metadata file.

    • mapperSpecificationKeys defines the routing rules after conversion.

    • serviceType identifies a field in the JSON returned by the adapter. ESP uses this field’s value to select the mapping specification.

    • external-event-simple-flat-usage identifies the selected mapping specification file in baseMappingDir.

    For a CSV event, Mapper Framework matches the incoming content type against supportedRecordTypes, invokes PartnerCsvConverter, and receives the converted JSON. ESP then uses the serviceType value in that JSON to select external-event-simple-flat-usage.yaml for BRM mapping.

    Note:

    Configure jsonConverters only when the incoming format requires conversion to JSON. For JSON input, ESP directly applies the routing and mapping specification rules.

  3. Save and close the application.yaml file.

Configuring Mapper Scenarios in Cloud Native Systems

In cloud native systems, configure the mapper’s application.yaml file by updating the configmap_esp_properties.yaml template in the BRM Helm chart.

To configure application.yaml in a cloud native system:

  1. Open the oc-cn-helm-chart/templates/configmap_esp_properties.yaml file in a text editor.

  2. In the application.yaml configuration, update the entries that apply to the incoming event.

  3. Save and close the file.

  4. Perform a Helm upgrade to apply the changes. For more information, see "Deploying BRM Cloud Native Services" in BRM Cloud Native Deployment Guide.

Configuring Mapper Specifications

A mapping specification is a YAML file that defines how ESP transforms an incoming event into BRM table records. ESP selects the mapping specification by using the routing rules in application.yaml.

A mapping specification contains the following entries:

  • mappings identifies the BRM tables to populate.

  • Each table contains fieldMappings.

  • Each fieldMappings entry defines the source, transformation, type, and target value for one BRM field.

To configure a mapping specification:

  1. Open the mapping specification file selected by application.yaml.

    Note:

    The directory specified by baseMappingDir contains the file.

    For information about the supplied mapping specifications and field mappings for ECE Composable Services TMF635 v5 rated events, see "About ECE Composable Services TMF635 v5 Mapping Specifications".

  2. Under mappings, add an entry for each BRM table that ESP must populate for the event.

  3. Under each table’s fieldMappings, add the BRM fields to populate. For information about the supported mapping approaches, see "Choosing a Mapping Approach". For a direct mapping, specify:

    • jsonFieldPath: Specifies the source field path in the incoming JSON event.

    • type: Specifies the required BRM field value type.

    For example, the following specification maps eventId and chargeAmount from an external event: 

    mappings:
  event_t:
    fieldMappings:
      event_no:
        jsonFieldPath: eventId
        type: STRING
  event_bal_impacts_t:
    fieldMappings:
      amount:
        jsonFieldPath: chargeAmount
        type: DOUBLE

    In this example, ESP creates mapped values for the event_t and event_bal_impacts_t BRM tables.

    Configure all tables and fields required for the event. For fields that require a default value, validation, value translation, transformation, an intermediate value, or custom Java logic, see "Customizing Mapper Specifications".

  4. Save and close the mapping specification file.

About ECE Composable Services TMF635 v5 Mapping Specifications

ESP provides the tmf635v5-produsage-gsm.yaml and tmf635v5-produsage-gprs.yaml mapping specification files for ECE Composable Services TMF635 v5 rated events. These files define the mandatory and optional ECE Composable Services field mappings and the transformations required by the TMF635 v5 contract.

For information about selecting the mapping specification for an incoming event, see "Configuring Mapper Scenarios in application.yaml".

The supplied specifications map BRM POIDs to the corresponding BRM reference fields. Table 7-2 lists the source fields for references in event_t.

Table 7-2 ECE Composable Services Reference Mappings for event_t

BRM Reference Source in the ECE Composable Services Payload Target Field Group

Account

relatedParty[*].partyOrPartyRole.id

ACCOUNT_OBJ_*

Service

product.id

SERVICE_OBJ_*

Item

usageProductPrice[0].itemRef.id

ITEM_OBJ_*

For references in event_bal_impacts_t, the supplied specifications use the corresponding usageProductPrice entry. The source paths in Table 7-3 are relative to usageProductPrice[*].

Table 7-3 ECE Composable Services Reference Mappings for event_bal_impacts_t

BRM Reference Source Within a Price Entry Target Field Group

Account

accountRef.id

ACCOUNT_OBJ_*

Balance group

balanceRef.id

BAL_GRP_OBJ_*

Item

itemRef.id

ITEM_OBJ_*

Purchased offering

purchasedProductOfferingRef.id

OFFERING_OBJ_*

Product

productOfferingExternalRef.id

PRODUCT_OBJ_*

Note:

In the target field names, * represents one of the following suffixes:

  • DB: Database number.

  • ID0: Object identifier.

  • REV: Revision.

  • TYPE: Object type.

Each POID component maps to its corresponding BRM column.

Table 7-4 lists the components and values of the example POID 0.0.0.1 /account 85081 0.

Table 7-4 BRM POID Components

Component Value

Database number

1

Object identifier

85081

Revision

0

Object type

/account

ECE Composable Services supplies PIN_FLD_ITEM_OBJ through itemRef.id. The supplied specifications map the event-level ITEM_OBJ columns from the first usageProductPrice entry. For each balance impact, they map ITEM_OBJ from itemRef in the corresponding price entry.

ECE Composable Services provides serialized BRM invoice data in the top-level invoiceData field of the ProductUsage JSON. ESP uses the supplied value and does not derive invoice data from balance impacts.

Table 7-5 shows the invoice-data mapping.

Table 7-5 ECE Composable Services Invoice Data Mapping

Source Field BRM Field Mapped Column Value Type

invoiceData

PIN_FLD_INVOICE_DATA

event_t.invoice_data

STRING

The following example shows the invoiceData field: 

{
  "invoiceData": "@INTEGRATE#0048100700#9731066455##0###<1 /item/misc 21060292 0#840#600##10.00#16384|1 /item/misc 21060292 0#840#600##-1.00#128|>"
}

The following mapping specification maps invoiceData to event_t.invoice_data: 

mappings:
  event_t:
    fieldMappings:
      invoice_data:
        jsonFieldPath: invoiceData
        type: STRING

Understanding Mapping Specification Attributes

You can use a mapping specification to define the BRM tables and fields that ESP populates for an event. Use mappings to define the BRM tables and fieldMappings to define the value for each target field.

For example, the following specification maps eventId from the incoming event to event_t.event_no: 

mappings:
  event_t:
    fieldMappings:
      event_no:
        jsonFieldPath: eventId
        type: STRING

Table 7-6 describes the entries and attributes in a mapping specification.

Table 7-6 Mapping Specification Entries and Attributes

Name Description When to Use
mappings Specifies the BRM tables to populate. Add an entry under mappings for each BRM table required for the event.
table-name Specifies a target BRM table, such as event_t or event_bal_impacts_t. Use the table name as the entry name under mappings.
fieldMappings Specifies the BRM fields to populate for a table. Add this entry under each target table.
field-name Specifies a target BRM field, such as event_no or amount. Add an entry under fieldMappings for each field to populate.
type Converts the resolved value to the required target type. Use when the target BRM field requires a specific value type, such as STRING, LONG, DOUBLE, or DECIMAL.
jsonFieldPath Specifies the source field path in the incoming JSON event. Supports nested paths and array indexes.

Use for a direct source-to-target mapping.

For information about the supported mapping approaches, see "Choosing a Mapping Approach".

defaultValue Specifies a value to use when the incoming event does not provide one.

Use for mandatory BRM fields that require a fixed default value.

For information about the supported mapping approaches, see "Choosing a Mapping Approach".

transformExp Specifies an expression that calculates, transforms, filters, combines, or translates input values by using the Mapper Framework generic-expression syntax.

Use when a direct jsonFieldPath mapping cannot produce the required value.

For information about the supported mapping approaches, see "Choosing a Mapping Approach".

inputFieldValidation Validates the source value before mapping or transformation. Use when the source value must satisfy a type, null, range, pattern, or other supported validation rule.
resultFieldMapping Translates a resolved value to a target value by using a configured value mapping. Use when an input code or value requires translation to a BRM-specific value.
tempField Makes a mapped value available for intermediate processing and reuse by later field mappings. RE Manager does not store the field in the final BRM table. Set this to true when multiple target fields use the same intermediate value.

For example, the following specification uses direct mappings, input validation, a transformation, and a default value:

mappings:
  event_t:
    fieldMappings:
      event_no:
        jsonFieldPath: eventId
        type: STRING

      charge_amount:
        jsonFieldPath: chargeAmount
        type: DOUBLE
        inputFieldValidation: "type:[number]"

      tax_amount:
        transformExp: transform(chargeAmount.multiply(taxRate))
        type: DOUBLE

      event_status:
        jsonFieldPath: status
        defaultValue: PENDING
        type: STRING

Note:

Use direct jsonFieldPath mappings whenever possible. Use additional attributes when a field requires validation, a default value, translation, transformation, or reuse of an intermediate value.

For information about selecting a mapping approach, see "Choosing a Mapping Approach".

For information about customizing field mappings, see "Customizing Mapper Specifications".

For information about implementing custom Java logic, see "Configuring Custom Java Transformations".

Choosing a Mapping Approach

You must use the simplest mapping approach that produces the required BRM value. Start with direct mapping, and use default values, value translation, generic expressions, or custom Java logic when the event requires them.

Table 7-7 describes when to use each mapping approach.

Table 7-7 Mapping Approaches

Mapping Approach When to Use Entries and Attributes
Direct mapping A source event field provides the value for a target BRM field directly. jsonFieldPath
Default-value mapping A target field requires a fixed value when the incoming event does not provide one. defaultValue
Value translation An input value requires translation to a BRM-specific code or value. jsonFieldPath and resultFieldMapping
Generic-expression mapping A target value requires calculations, conditional processing, aggregation, filtering, string processing, date conversion, or multiple source fields. transformExp
Intermediate-value mapping Multiple mappings reuse a calculated value that RE Manager must not store in the final BRM table. transformExp, tempField

Reusing a mapped field

You can reuse a value calculated by an earlier field mapping in two ways.

  • Reuse Without Additional Transformation: Use jsonFieldPath to copy the previously mapped value directly.
    amount:
  jsonFieldPath: event_t.normalized_amount
  type: DOUBLE
  • Reuse With Additional Transformation: Use transformExp when further calculation or conversion is required.
    rounded_amount:
  transformExp: transform(field.event_t.normalized_amount.round())
  type: DOUBLE

Note: Define the reusable field before the fields that use it. Use the table-qualified field name to avoid a collision with an input JSON field of the same name.

First, define the reusable value.

Note: Use tempField: true when it is only an intermediate value and must not be included in the final BRM output. 

mappings:
  event_t:
    fieldMappings:
      normalized_amount:
        transformExp: transform(root.chargeAmount.divide(100))
        tempField: true
        type: DOUBLE

Use the following formats to reference input JSON fields and previously mapped fields:

  • Read an input JSON field: 
    jsonFieldPath: chargeAmount
  • Copy a previously mapped field: 
    jsonFieldPath: event_t.normalized_amount
  • Transform a previously mapped field:
    transformExp: transform(field.event_t.normalized_amount)
Validated mapping A source value must satisfy validation rules before mapping or transformation. jsonFieldPath or transformExp, inputFieldValidation
Custom Java mapping Standard mapping attributes and generic expressions cannot implement the required business rule. transformClass, transformExp, and a custom transformation method

Note:

The type attribute is optional for all mapping approaches.

The following topics describe these mapping approaches:

Configuring Direct Mappings

You can use direct mapping when one source field maps directly to one BRM field.

To configure a direct mapping:

  1. Open the mapping specification file selected by application.yaml.

  2. Under the target table’s fieldMappings entry, add or update the target BRM field and configure the following:

    • jsonFieldPath: Specifies the source field path in the incoming JSON event.

    • type: Specifies the required BRM field value type.

    For example, the following mapping reads eventId from the incoming event and maps it to event_no: 

    event_no:
  jsonFieldPath: eventId
  type: STRING

  3. Save and close the mapping specification file.

Configuring Default-Value Mappings

You can use defaultValue to supply a value for a target BRM field when the incoming event does not provide one.

To configure a default-value mapping:

  1. Open the mapping specification file selected by application.yaml.

  2. Under the target table’s fieldMappings entry, add or update the target BRM field and configure the following:

    • jsonFieldPath: Specifies the source field path, if the incoming event can provide the value.

    • defaultValue: Specifies the value to use when the incoming event does not provide one.

    • type: Specifies the required BRM field value type.

    For example, the following mapping uses PENDING when the incoming event does not provide a value for status: 

    status:
  jsonFieldPath: status
  defaultValue: PENDING
  type: STRING

  3. Save and close the mapping specification file.

Configuring Value Translation

You can use resultFieldMapping to translate an incoming value to a BRM-specific code or value.

To configure value translation:

  1. Open the mapping specification file selected by application.yaml.

  2. Under the target table’s fieldMappings entry, add or update the target BRM field and configure the following:

    • jsonFieldPath: Specifies the source field path in the incoming JSON event.

    • resultFieldMapping: Specifies the source values and their corresponding target values.

    • type: Specifies the required BRM field value type.

    For example, the following mapping translates usageType values to numeric values for resource_type: 

    resource_type:
  jsonFieldPath: usageType
  resultFieldMapping: "[(VOICE,1), (DATA,2), (SMS,3)]"
  type: LONG

    In this example, the mapping translates VOICE to 1, DATA to 2, and SMS to 3.

  3. Save and close the mapping specification file.

Configuring Generic-Expression Mappings

You can use transformExp when a target value requires transformation or calculation.

Generic expressions support value selection, arithmetic, conditional processing, filtering, aggregation, string processing, date conversion, and access to previously calculated temporary fields.

To configure a generic-expression mapping:

  1. Open the mapping specification file selected by application.yaml.

  2. Under the target table’s fieldMappings entry, add or update the target BRM field and configure the following:

    • transformExp: Specifies the expression that calculates or transforms the value.

    • type: Specifies the required BRM field value type.

    For example:

    end_time:
  transformExp: transform(start_time.add(15))
  type: LONG

  3. Save and close the mapping specification file.

For examples of supported generic-expression capabilities, review the installed BRM_home/apps/event_stream_processor/spec/generic-expression-reference.yaml file.

This reference specification demonstrates direct mapping, default values, temporary fields, input validation, result-field mapping, cross-field processing, transformations, and reusable expressions.

Configuring Intermediate-Value Mappings

You can use a temporary field when multiple BRM fields use the same intermediate result. Later field mappings can reference the temporary value. RE Manager does not store the temporary field in the final BRM table.

To configure an intermediate-value mapping:

  1. Open the mapping specification file selected by application.yaml.

  2. Under the table’s fieldMappings entry, add a field for the intermediate value and configure the following:

    • transformExp: Specifies the expression that calculates the intermediate value.

    • tempField: Set this to true to make the value available for intermediate processing.

    • type: Specifies the intermediate value type.

    For example:

    normalized_amount:
  transformExp: transform(chargeAmount.divide(100))
  tempField: true
  type: DOUBLE

  3. Reference the temporary value in a later field mapping.

    For example:

    amount:
  transformExp: transform(field.event_t.normalized_amount)
  type: DOUBLE

    The following examples show alternative references to the temporary value:

    transformExp: event_t.normalized_amount
    transformExp: normalized_amount

  4. Save and close the mapping specification file.

About Custom Java Mappings

You can use a custom Java transformation when standard mapping attributes and generic expressions cannot implement the required business rule.

Keep custom logic focused on that rule. Use application.yaml for routing and the mapping specification for field selection and ordinary transformations.

For information about implementing and deploying a custom transformer, see "Configuring Custom Java Transformations".

Customizing Mapper Specifications

You can customize a mapping specification to supply default values, translate or transform source values, and make values available for intermediate processing.

You can use the following entries under a field’s fieldMappings entry:

  • defaultValue: Specifies a value to use when the incoming event does not provide one.

    Note:

    If a field has no mapped value and no defaultValue is configured, ESP uses a default based on the field’s data type when committing the event. For example, the default for an integer field is 0.

  • resultFieldMapping: Maps an incoming value to the corresponding BRM value.

  • transformExp: Specifies an expression that calculates or transforms a value.

  • tempField: Set this to true to make a field available for intermediate processing, such as event enrichment. RE Manager does not store the field in the final BRM table.

To customize a mapping specification:

  1. Open the specification file in the directory specified by baseMappingDir.

  2. Add or update the field mappings that the event requires. If the transformation uses a transformer class, set transformClass at the beginning of the specification file.

    For example, the following mappings from external-event-simple-flat-usage.yaml make login_id available for intermediate processing, convert eventTime, and map currency codes to BRM balance element IDs. Keep the other mappings required for the event in the same file. 

    transformClass: com.oracle.brm.mapper.hook.customFieldTransformer

mappings:
  event_t:
    fieldMappings:
      login_id:
        jsonFieldPath: serviceLoginId
        type: STRING
        tempField: true
      start_t:
        transformExp: transform(eventTime.epoch())
        type: LONG
  event_bal_impacts_t:
    fieldMappings:
      resource_id:
        jsonFieldPath: currency
        resultFieldMapping: "[(INR,356),(USD,840),(EUR,978)]"
        type: INTEGER

    The following example shows how a different external event specification uses defaultValue to populate descr: 

    mappings:
  event_t:
    fieldMappings:
      descr:
        defaultValue: Prerated telecom voice usage
        type: STRING

  3. Save and close the file.

Deploying Custom Mapper Specifications in Cloud Native Systems

In cloud native systems, you can deploy custom ESP mapping specification files by using a Kubernetes ConfigMap without rebuilding the ESP image.

Table 7-8 lists the locations and configuration entry used for specification deployment.

Table 7-8 Mapper Specification Deployment Locations and Settings

Item Location or Value

Helm chart directory for supplied specifications

esp_specs/

ConfigMap for supplied specifications

esp-specs

Runtime specification directory

/oms/apps/event_stream_processor/eem/scenarios/spec

Custom ConfigMap configuration entry

ocbrm.event_stream_processor.extEspCustomSpecsCM

To deploy custom mapping specification files:

  1. Copy the custom .yaml mapping specification files to a local directory.

  2. Create a ConfigMap from the directory. For example: 

    kubectl -n <namespace> create configmap esp-custom-specs \
  --from-file=<spec_directory>/

    where:

    • namespace is the BRM Kubernetes namespace.

    • spec_directory is the directory that contains the custom mapping specification files.

  3. Set ocbrm.event_stream_processor.extEspCustomSpecsCM to the ConfigMap name in the deployment values file. For example: 

    ocbrm:
  event_stream_processor:
    extEspCustomSpecsCM: esp-custom-specs

  4. In application.yaml, configure mapperSpecificationKeys to select the custom mapping specification file. For more information, see "Configuring Mapper Scenarios in application.yaml".

  5. Deploy the updated Helm values. For more information, see "Deploying BRM Cloud Native Services" in BRM Cloud Native Deployment Guide.

Note:

To override a supplied mapping specification, use the same file name in the custom ConfigMap. ESP copies the supplied files first and the custom files next. If the file names match, ESP uses the custom file.

Updating Supplied Mapper Specifications

In cloud native systems, the supplied ESP mapping specification files are located in the oc-cn-helm-chart/esp_specs directory.

To update a supplied mapping specification:

  1. Open the mapping specification file in the oc-cn-helm-chart/esp_specs directory in a text editor.

  2. Update the mappings as required for your events. For more information, see "Customizing Mapper Specifications".

  3. Save and close the file.

  4. Perform a Helm upgrade to apply the changes. For more information, see "Deploying BRM Cloud Native Services" in BRM Cloud Native Deployment Guide.

Configuring Custom Java Transformations

You can implement a custom Java transformation when direct mapping, value translation, validation, temporary fields, and generic expressions cannot implement the required business rule.

Configure the transformer class in the mapping specification and invoke its methods through transformExp.

The following topics describe how to implement and use a custom transformer:

Configuring a Custom Transformer Class

You can use transformClass to specify the custom transformer class that contains the methods referenced by transformExp.

To configure a custom transformer class:

  1. Open the mapping specification file selected by application.yaml.

  2. At the beginning of the file, set transformClass to the fully qualified name of your transformer class: 

    transformClass: fully-qualified-class-name

    For example:

    transformClass: com.example.mapper.PartnerFieldTransformer

  3. Save and close the mapping specification file.

Custom methods that you reference from transformExp must:

  • Be public.

  • Have names that start with fn.

  • Accept a Map<String, Object> argument and a JsonFieldContext argument.

  • Return the mapped value as Object.

For an implementation example, see "Implementing a Custom Transformation Method".

Implementing a Custom Transformation Method

You can implement a custom transformation method to process a source value and return the value required by a BRM field.

The following example reads customerId from the incoming event, optionally adds a prefix supplied by the mapping specification, and returns a normalized value. 

package com.example.mapper;

import java.util.List;
import java.util.Map;

import com.fasterxml.jackson.databind.JsonNode;
import com.oracle.brm.mapper.model.JsonFieldContext;
import com.oracle.brm.mapper.util.JsonPathUtil;
import com.oracle.brm.mapper.util.MapperConstants;

public class PartnerFieldTransformer {

    public Object fnNormalizeCustomerId(
            Map<String, Object> args,
            JsonFieldContext context) {

        JsonNode payload = (JsonNode) args.get(MapperConstants.PAYLOAD);
        String jsonPath = (String) args.get(MapperConstants.JSON_PATH);

        @SuppressWarnings("unchecked")
        List<Object> extraArgs =
                (List<Object>) args.get(MapperConstants.EXTRA_ARGS);

        String customerId =
                JsonPathUtil.getValueAsStringFromJnode(
                        context, payload, jsonPath);

        if (customerId == null || customerId.isBlank()) {
            return "";
        }

        String prefix = extraArgs == null || extraArgs.isEmpty()
                ? ""
                : String.valueOf(extraArgs.get(0));

        return prefix + customerId.trim().toUpperCase();
    }
}

In this example, the fnNormalizeCustomerId method returns an empty string when customerId is null or blank. Otherwise, it removes leading and trailing spaces, converts the customer ID to uppercase, and adds the optional prefix.

For information about calling the method from a mapping specification, see "Invoking a Custom Transformation Method".

Invoking a Custom Transformation Method

You can use transformExp to invoke a method in the class specified by transformClass.

To invoke a custom transformation method:

  1. Open the mapping specification file selected by application.yaml.

  2. Set transformClass at the beginning of the file to the fully qualified name of your transformer class.

  3. Under the target table’s fieldMappings entry, configure the target BRM field:

    • transformExp: Specifies the source JSON path, custom method name, and any additional arguments.

    • type: Specifies the required BRM field value type.

    For example:

    transformClass: com.example.mapper.PartnerFieldTransformer

mappings:
  event_t:
    fieldMappings:
      customer_id:
        transformExp: transform(customerId, fnNormalizeCustomerId, "CUST-")
        type: STRING

    In this example:

    • customerId specifies the source JSON path.

    • fnNormalizeCustomerId specifies the public method in PartnerFieldTransformer.

    • The framework passes "CUST-" to the method through MapperConstants.EXTRA_ARGS.

    • ESP maps the result to event_t.customer_id.

  4. Save and close the mapping specification file.

About Custom Transformation Method Arguments

You can access the values that the mapper framework supplies to your custom transformation method through the args map.

Table 7-9 describes the available argument constants.

Table 7-9 Custom Transformation Method Arguments

Constant Description
MapperConstants.PAYLOAD The root incoming event as a JsonNode.
MapperConstants.JSON_PATH The source JSON path specified in transformExp.
MapperConstants.CONTEXT The current mapping context, when available.
MapperConstants.INDEX The current array-row index when mapping repeated data.
MapperConstants.BRM_TABLE_NAME The current target BRM table name.
MapperConstants.BRM_FIELD_NAME The current target BRM field name.
MapperConstants.RESULT_FIELD_MAPPING The configured resultFieldMapping value, when present.
MapperConstants.EXTRA_ARGS Additional arguments specified after the custom method name in transformExp.

For example:

transformExp: transform(product.id, fnExtractValue, "id", "numeric")

In this example, the custom method receives "id" and "numeric" through MapperConstants.EXTRA_ARGS.

Compiling and Deploying a Custom Transformer

You can compile your custom transformer against the BRM libraries and deploy its JAR file on the ESP Java classpath.

To compile and deploy a custom transformer:

  1. Compile the transformer against the BRM libraries:

    mkdir -p build/classes

javac -cp "$PIN_HOME/jars/*" \
  -d build/classes \
  source-file-path

    where source-file-path is the path to your transformer’s Java source file.

    For example:

    javac -cp "$PIN_HOME/jars/*" \
  -d build/classes \
  src/com/example/mapper/PartnerFieldTransformer.java

  2. Package the compiled class in a JAR file:

    jar cf jar-file-name -C build/classes .

    where jar-file-name is the name of the JAR file to create.

    For example:

    jar cf partner-mapper-customizations.jar -C build/classes .

  3. Copy the JAR file to the customer-managed ESP extension location.

  4. Add the JAR file to the ESP Java classpath before starting ESP.

After deploying a new version of the JAR file, restart ESP so that it reloads the class specified by transformClass.

Creating a Custom Format Adapter

You can create a custom format adapter to process incoming events that use a non-JSON format. The adapter converts one input record to a Jackson JsonNode. The mapper framework then applies the JSON routing and mapping rules.

The mapper framework provides the JsonConverter plug-in interface. You must provide:

  • The format-specific parsing logic.

  • Any format-specific metadata files.

  • Any additional compile-time and runtime dependencies.

  • The custom adapter JAR file on the ESP Java classpath.

The following topics describe how to implement and use a custom format adapter:

Note:

You must keep the adapter focused on converting the incoming record to JSON. Configure routing in application.yaml. Configure validation, default values, transformations, and BRM table and field mappings in the selected mapping specification.

About the Format Adapter Interface

You can implement the com.oracle.brm.mapper.core.spi.JsonConverter interface to create a custom format adapter.

Your adapter class must:

  • Implement com.oracle.brm.mapper.core.spi.JsonConverter.

  • Be public.

  • Have a public no-argument constructor.

  • Return a non-null JsonNode from its convert method.

  • Throw MapperException when it cannot complete the conversion.

Implement the convert method in your adapter class to convert the incoming record to JSON: 

JsonNode convert(Object record, JsonConverterContext context)
        throws MapperException;

where:

  • record contains the incoming record to convert.

  • context provides access to the adapter configuration and incoming record type.

The record argument can contain a supported input object, such as a string, byte array, input stream, or CloudEvent. Use JsonNodeExtractor.textFromObject() to read the input as text.

Converting Delimited Text to JSON

You can implement a custom format adapter to convert a delimited input record to the JSON fields required by a mapping specification.

The following example uses a pipe-delimited record that contains an event ID, service type, and charge amount:

EVT-1001|/service/telco/gsm/telephony|12.50

The adapter reads the delimiter and event scenario from its configuration:

package com.example.mapper;

import java.util.regex.Pattern;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;

import com.oracle.brm.mapper.core.JsonNodeExtractor;
import com.oracle.brm.mapper.core.MapperErrorCode;
import com.oracle.brm.mapper.core.MapperException;
import com.oracle.brm.mapper.core.spi.JsonConverter;
import com.oracle.brm.mapper.core.spi.JsonConverterContext;

public final class PartnerDelimitedAdapter implements JsonConverter {

    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();

    public PartnerDelimitedAdapter() {
    }

    @Override
    public JsonNode convert(Object record, JsonConverterContext context)
            throws MapperException {

        String input = JsonNodeExtractor.textFromObject(record);
        if (input == null || input.isBlank()) {
            throw MapperException.of(
                    MapperErrorCode.MF_CONVERSION_FAILED,
                    "Partner input record cannot be empty");
        }

        String delimiter = context.getText("delimiter");
        if (delimiter == null || delimiter.length() != 1) {
            throw MapperException.of(
                    MapperErrorCode.MF_CONVERSION_FAILED,
                    "Adapter configuration 'delimiter' must contain one character");
        }

        String[] values = input.trim().split(
                Pattern.quote(delimiter), -1);

        if (values.length != 3) {
            throw MapperException.of(
                    MapperErrorCode.MF_CONVERSION_FAILED,
                    "Partner input record must contain event ID, service type, and amount");
        }

        ObjectNode result = OBJECT_MAPPER.createObjectNode();
        result.put("eventId", values[0].trim());
        result.put("serviceType", values[1].trim());
        result.put("chargeAmount", values[2].trim());
        result.put("eventScenario", context.getRequiredText("eventScenario"));

        return result;
    }
}

The adapter returns the following JSON:

{
  "eventId": "EVT-1001",
  "serviceType": "/service/telco/gsm/telephony",
  "chargeAmount": "12.50",
  "eventScenario": "PARTNER_USAGE"
}

For information about configuring the delimiter, event scenario, and routing rules for this adapter, see "Configuring a Custom Format Adapter".

Configuring a Custom Format Adapter

You can configure your custom format adapter class and its settings in application.yaml. The adapter defines the settings under configuration, and the framework passes them through JsonConverterContext.

To configure a custom format adapter:

  1. Open the application.yaml file at the path specified by mapper.configuration.path in the ESP Infranet.properties file.

  2. Under jsonConverters, add an entry for your adapter and configure the following:

    • supportedRecordTypes: Specifies the incoming record types that the adapter handles.

    • converterClass: Specifies the fully qualified class name of your adapter.

    • configuration: Specifies the settings that your adapter requires.

    Use the following format for converterClass: 

    converterClass: fully-qualified-adapter-class-name

  3. Configure mapperSpecificationKeys to select the mapping specification for the converted event.

    For example:

    jsonConverters:
  partner-delimited:
    supportedRecordTypes:
      - application/vnd.partner.usage
    converterClass: com.example.mapper.PartnerDelimitedAdapter
    configuration:
      delimiter: "|"
      eventScenario: PARTNER_USAGE

mapperSpecificationKeys:
  eventScenario:
    PARTNER_USAGE: partner-usage

    where:

    • partner-delimited identifies the adapter configuration.

    • delimiter specifies the character that separates the input values.

    • eventScenario specifies the scenario value that the adapter adds to the converted event.

    • partner-usage identifies the partner-usage.yaml mapping specification in the directory specified by baseMappingDir.

    Keep the other configuration entries required for the event in the same file.

  4. Save and close the application.yaml file.

Table 7-10 describes the context methods that your adapter can use.

Table 7-10 Format Adapter Context Methods

Method Description
getRecordType() Returns the matched incoming record type.
getText(name) Returns an optional textual configuration value.
getRequiredText(name) Returns a required textual configuration value.
resolvePath(name) Resolves a configured file path relative to application.yaml.
getConfiguration() Returns a defensive copy of the complete adapter configuration as a JsonNode.

For information about deploying the adapter JAR file and restarting ESP, see "Compiling and Deploying a Custom Format Adapter".

Compiling and Deploying a Custom Format Adapter

You can compile your custom format adapter against the BRM libraries and any additional dependencies that it requires. Deploy the adapter JAR file and its runtime dependencies on the ESP Java classpath.

To compile and deploy a custom format adapter:

  1. Compile the adapter against the BRM libraries. Include any additional compile-time dependencies that your adapter requires.

    For an adapter that requires only the BRM libraries, use:

    mkdir -p build/classes

javac -cp "$PIN_HOME/jars/*" \
  -d build/classes \
  source-file-path

    where source-file-path is the path to your adapter’s Java source file.

    For example:

    javac -cp "$PIN_HOME/jars/*" \
  -d build/classes \
  src/com/example/mapper/PartnerDelimitedAdapter.java

  2. Package the compiled class in a JAR file:

    jar cf jar-file-name -C build/classes .

    where jar-file-name is the name of the JAR file to create.

    For example:

    jar cf partner-format-adapter.jar -C build/classes .

  3. Add the adapter JAR file and any required runtime dependencies to the ESP Java classpath.

  4. Configure converterClass in application.yaml to specify your adapter class. See "Configuring a Custom Format Adapter".

  5. Restart ESP.

About Event Enrichment

ESP uses enrichment classes to add BRM information that an external rated event does not contain. For example, an event can contain a rated amount but lack the item type, service object, or invoice data required for loading. ESP runs the configured classes on mapped events before it stores them in the BRM database.

Enrichment depends on the event source:

  • For internal events: ECE REF JSONFILE events use the internal processing path. Do not configure the external event enrichment classes for the internal event handler.

  • For ECE Composable Services events: The delivered TMF processing path does not require the external event batch enrichment classes.

  • For external events: Configure the enrichment classes needed to populate missing BRM information, such as item types, tax codes, general ledger IDs (G/L IDs), account and service objects, balance impacts, and invoice data. See Table 5-20 for more information.

For more information about configuring the delivered enrichment classes, see "Configuring Event Enrichment".

For more information about adding enrichment logic that these classes do not provide, see "Configuring a Custom Enrichment Class".

Configuring Event Enrichment

ESP provides the following selector classes for external rated events. These classes use rules in the generic_selector.yaml file:

  • TaxSelector uses the configured rules to select a tax code.

  • ItemSelector uses the configured rules to select an item type. If no rule matches, it selects /item/misc.

  • GlidSelector uses the configured rules to select a G/L ID when the event does not supply one.

ESP also provides the following enrichment classes:

  • BalanceImpactProcessor uses the mapped event to prepare balance-impact information.

  • ServiceItemLookup uses the login identifier, event time, service type, and item type to find the account, service, balance group, and item.

  • InvoiceDataPopulator uses information from the event and its balance impacts to populate invoice data.

You can use a custom policy hook to apply customer-specific logic to external rated events after invoice-data generation and before ESP commits them.

Before configuring enrichment, configure the Kafka topic and event handler for external rated events. The mapper specification must supply the fields required by the classes you use. For example, ServiceItemLookup needs a login identifier, event time, service type, and item type. ItemSelector can supply the item type. See "Configuring Kafka Consumer, Topics, and Event Handlers" and "Configuring Mapper Specifications".

If you use a selector, configure its rules before registering the enrichment classes:

Configuring Selector Rules

Configure the rules for the selectors you use and retain the other sections in the file.

Note:

In BRM cloud native systems, the generic_selector.yaml file is included in the ESP image and cannot be edited directly.

To configure selector rules in a cloud native system:

  1. Add a generic_selector.yaml file containing your selector rules to the oc-cn-helm-chart/esp_specs directory. Retain the other selector sections in the file.

  2. Open the oc-cn-helm-chart/templates/configmap_esp_properties.yaml file in a text editor.

  3. Set event_stream_processor.selector.config_path to the following location: 

    /oms/apps/event_stream_processor/eem/scenarios/spec/generic_selector.yaml

  4. Save and close the file.

  5. Perform a Helm upgrade to apply the changes. For more information, see "Deploying BRM Cloud Native Services" in BRM Cloud Native Deployment Guide.

To configure selector rules on an on-premises system:

  1. Open BRM_home/apps/event_stream_processor/generic_selector.yaml in a text editor.

  2. Under configObjects, configure the applicable event information, rule order, conditions, and result for each selector you use.

    The following example shows an ItemSelector rule that selects an item specification for a GSM event with a balance impact on resource 840: 

    configObjects:
  itemTypeSelectors:
    - name: GSM Delayed Session Item Selector
      description: Item selector for delayed GSM session events
      applicableToName: TelcoGSMTelephony
      eventSpecName: DelayedSessionTelcoGSM
      applicableToAllBalImpact: false
      rules:
        - ruleName: Usage impact for resource 840
          ruleOrder: 1
          conditions:
            - fieldName: EVENT_T.POID_TYPE
              operation: EQUAL_TO
              fieldValue: /event/delayed/session/telco/gsm
            - fieldName: EVENT_T.SERVICE_OBJ_TYPE
              operation: EQUAL_TO
              fieldValue: /service/telco/gsm/telephony
            - fieldName: EVENT_BAL_IMPACTS_T.IMPACT_TYPE
              operation: EQUAL_TO
              fieldValue: 1
            - fieldName: EVENT_BAL_IMPACTS_T.RESOURCE_ID
              operation: EQUAL_TO
              fieldValue: 840
          specName: Item-gsm-usage

    Note:

    Replace the sample event attributes, conditions, and item specification with values for your BRM system. Retain the other selector sections in the generic_selector.yaml file.

  3. Save and close the file.

Configuring Enrichment Classes

ESP runs enrichment classes in the order specified in the Infranet.properties file. Add only the classes required by your events. Place each class after any class that supplies its input.

To configure enrichment classes:

  1. Open the BRM_home/apps/event_stream_processor/Infranet.properties file in a text editor.

  2. If you use selectors, set event_stream_processor.selector.config_path to the location of generic_selector.yaml. See Table 5-20.

  3. Add a numbered enrichment entry for each class your events require. See Table 5-20.

    Configure TaxSelector or GlidSelector and their rules if your events require them.

  4. Save and close the file.

  5. Start or restart ESP.

Configuring a Custom Enrichment Class

You can add a custom enrichment class to supply BRM information that the mapped external event and delivered enrichment classes do not provide. Your class must implement BatchProcessor and its handleBatch(...) method. ESP runs the class on batches of mapped events in the order you specify in the Infranet.properties file.

To configure a custom enrichment class:

  1. Create a Java class that implements BatchProcessor. In handleBatch(...), populate the required fields in the mapped events.

  2. Install the class so that ESP can load it.

  3. Open BRM_home/apps/event_stream_processor/Infranet.properties in a text editor.

  4. Add the class to the numbered enrichment entries. Place it after the classes that supply its input and before any class that uses its output. See Table 5-20.

    For example, the following entries run a service and item lookup class, a custom class, and an invoice data class in that order:

    event_stream_processor.batch.enrichment.class.1=com.oracle.brm.esp.core.event_stream_processor.ServiceItemLookup
event_stream_processor.batch.enrichment.class.2=com.example.brm.CustomEnrichment
event_stream_processor.batch.enrichment.class.3=com.oracle.brm.esp.core.event_stream_processor.InvoiceDataPopulator

    where com.example.brm.CustomEnrichment is the name of your class.

  5. Save and close the file.

  6. Start or restart ESP.

Preventing Duplicate Event Processing

For ECE Composable Services and external rated events, ESP can check for duplicates before it stores events in the BRM database. ESP identifies an event by its event type and event key. If ESP has already accepted that combination in the current batch or an earlier batch, it marks the duplicate as a failed event. ESP retains accepted keys for the configured period.

To configure ESP to check for duplicate events:

  1. Open the BRM_home/apps/event_stream_processor/Infranet.properties file in a text editor.

  2. Configure ESP to check for duplicate events and set how many days it keeps event keys. See Table 5-18.

  3. Save and close the file.

  4. Start or restart ESP.

Monitoring ESP

To monitor ESP:

  1. Start ESP by running the start_event_stream_processor script in the BRM_home/bin directory.

  2. Check the ESP application log for startup, Kafka topic subscription, handler, mapper, and database connection errors. To change the log level, edit the BRM_home/apps/event_stream_processor/log4j2.xml file. See "Configuring the Log Level in Event Stream Processor".

  3. Check the metrics at http://host:port/observe/metrics for Kafka consumer activity, mapping results, and database commits. In cloud native deployments, you can also use the configured Grafana dashboard. For more information, see Table 5-10 and Table 5-11.

  4. Confirm that ESP stores accepted events as zipped JSON in the BRM database and that RE Manager loads them. If a batch stalls or events repeatedly fail, check the metrics and ESP log to identify the failure. For more information, see "Managing Failed Rated Events".

To stop ESP, run the stop_event_stream_processor script in the BRM_home/bin directory.

Managing Failed Rated Events

If mapping, enrichment, or duplicate-event checking fails for a rated event, ESP stores the failed event in a /suspended_rated_events object. The object contains the event payload and information about the failure, including the error code, error description, and status.

If ESP cannot commit a batch to the BRM database, it rolls back the batch and its database offset update. ESP’s Kafka consumer can then read the affected records again. Before recycling an event, check whether ESP has already stored it in BRM.

Configuring Failed Rated Event Recovery

Configure EAI JS and DM Kafka to publish recycled rated events to a Kafka topic that ESP consumes. Complete the following configuration before running the PCM_OP_SUSPENSE_RECYCLE_ALL_RATED_EVENTS opcode.

To configure failed rated event recovery:

  1. Add the following entry to the BRM event notification configuration:

    1301 0 /event/notification/suspense/recycle/rated_event

    Note:

    Retain all existing entries.

    • For an on-premises system, add the entry to BRM_home/sys/data/config/pin_notify_kafka_sync.

    • For a cloud native system, add the entry to the notification configuration in oc-cn-helm-chart/templates/configmap_pin_notify.yaml.

  2. In the EAI JS Kafka payload configuration file, add the RecycleRatedEvent publisher, Kafka header, and event entries to the existing <PublisherDefs>, <ElementDefs>, and <EventDefs> elements, respectively.

    The following example shows the entries:

    <PublisherDefs>
  <Publisher DB="eai_dm_database_number" Format="FLIST">
    RecycleRatedEvent
  </Publisher>
</PublisherDefs>
<ElementDefs>
  <KafkaHeaderInfo Source="EVENT"
                   PinFld="PIN_FLD_KAFKA_HEADER_INFO"
                   DataFrom="PIN_FLD_KAFKA_HEADER_INFO"
                   Tag="KafkaHeaderInfo">
    <Field PinFld="PIN_FLD_HEADER_INFO" Tag="HeaderInfo"/>
  </KafkaHeaderInfo>
</ElementDefs>
<EventDefs>
  <RecycleRatedEvent Source="EVENT"
                     Tag="RecycleRatedEvent"
                     StartEvent="/event/notification/suspense/recycle/rated_event">
    <Attribute Tag="Version" Value="1.0"/>
    <Field PinFld="PIN_FLD_PAYLOAD_STR" Tag="PayloadStr"/>
    <SubElement Name="KafkaHeaderInfo"
                OnEvent="/event/notification/suspense/recycle/rated_event"/>
  </RecycleRatedEvent>
</EventDefs>

    where eai_dm_database_number is the EAI DM database number for your system.

    Note:

    Retain the other definitions. Add the entries within the existing <PublisherDefs>, <ElementDefs>, and <EventDefs> elements. Do not duplicate these elements.

  3. Configure EAI JS to use the payload configuration file:

    • For an on-premises system, set infranet.eai.configFile in the EAI JS Infranet.properties file to the file’s location.

    • For a cloud native system, deploy the file with EAI JS and set eai_js.deployment.eaiConfigFile to its file name in the applicable values file.

  4. Configure DM Kafka to publish recycled rated events to a Kafka topic that ESP consumes.

    • For an on-premises system, configure the topic and its business event mapping in BRM_home/sys/dm_kafka/dm_kafka_config.xml. See "Mapping Business Events to Kafka Topics" in the BRM Developer’s Guide.

    • For a cloud native system, set the topic name in the applicable deployment values file:

      dm_kafka:
  deployment:
    topicName: topic_name
    topicFormat: JSON
    topicStyle: ShortName

      where topic_name is the Kafka topic configured for ESP to consume.

  5. Configure ESP to consume the topic specified in step 4 and process its recycled rated events:

    • For an on-premises system, configure the topic in BRM_home/apps/event_stream_processor/Infranet.properties and verify that its handler processes recycled rated events. For more information, see Table 5-12 and Table 5-13.

    • For a cloud native system, add or verify the topic in the ESP configuration. The following example uses the topic name from step 4: 

      topicGroups:
  - name: external
    topics:
      - name: "topic_name"

      Note:

      If ESP already consumes the topic, verify that its handler processes recycled rated events.

  6. Load the updated event notification configuration.

    • For an on-premises system, run the following command on the BRM host:

      load_pin_notify file_path

      where file_path is the path to the updated event notification configuration file.

    • For a cloud native system, deploy the updated configuration. From the sys/data/config directory in the PJE pod, run: 

      load_pin_notify -d pin_notify_kafka_sync

  7. Restart CM and EAI JS after applying their configuration changes. Apply any DM Kafka and ESP changes using the procedure for your deployment.

After completing the configuration, see "Reprocessing Failed Rated Events" for more information about running PCM_OP_SUSPENSE_RECYCLE_ALL_RATED_EVENTS.

Viewing Failed Rated Events

When ESP cannot process an individual rated event, it stores the event in a /suspended_rated_events object. Review the object’s error details to identify the cause before reprocessing it.

To view failed rated events:

  1. Use BRM Object Browser or a BRM object search to find /suspended_rated_events objects in the failed state.

  2. Open an object and review its error code, error description, status, and attempt count. Review the event payload if it is available.

  3. Check the ESP log for more information about the failure. For a mapping or enrichment error, compare the event with the applicable configuration. For a duplicate, compare its event type and event key with the event that ESP previously accepted.

Note:

If a database commit error causes ESP to roll back a batch, check the ESP log for the batch failure. The rollback also reverses the database offset update, so ESP's Kafka consumer can read the affected records again.

After correcting the cause of a failed event, see "Reprocessing Failed Rated Events".

Reprocessing Failed Rated Events

The PCM_OP_SUSPENSE_RECYCLE_ALL_RATED_EVENTS opcode recycles all suspended rated events in the failed state. Configure failed rated event recovery before running the opcode. For more information, see "Configuring Failed Rated Event Recovery".

To reprocess failed rated events:

  1. Review all failed events and correct the causes of their failures.

    Note:

    Run the opcode only when every event in the failed state is ready for reprocessing.

  2. Save the following testnap input flist in a file named recycle: 

    0 PIN_FLD_POID POID [0] database_number /suspended_rated_events 0 0

    where database_number is your BRM database number.

  3. Start testnap on the BRM host or, for a cloud native system, from the CM pod. Run the following commands: 

    r recycle 1
          xop PCM_OP_SUSPENSE_RECYCLE_ALL_RATED_EVENTS 0 1

  4. Check the /suspended_rated_events objects for updated statuses and error details.

    Note:

    After the event is published for reprocessing, its status changes to Recycle. It then changes to Success if processing succeeds or returns to Failed if processing fails again.