<commit_before><?xml version="1.0" encoding="utf-8"?>
<tnt-data-source
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="https://raw.githubusercontent.com/Nastel/tnt4j-streams/master/config/tnt-data-source.xsd">

    <!-- START of metrics parsing -->

    <parser name="MetricsParser" class="com.jkoolcloud.tnt4j.streams.parsers.ActivityMapParser">
        <field name="All" locator="*" locator-type="Label"/>
    </parser>

    <parser name="SendMetricsParser" class="com.jkoolcloud.tnt4j.streams.parsers.ActivityJsonParser" default-data-type="AsInput"
            manualFieldsOrder="true">
        <property name="ActivityDelim" value="EOF"/>

        <field name="topic" locator="$.Topic" locator-type="Label" required="true"/>
        <field name="OperationType" locator="$.Type" locator-type="Label" transparent="true">
            <field-map source="ProducerTopicMetrics" target="Producer"/>
            <field-map source="ConsumerTopicMetrics" target="Consumer"/>
        </field>

        <field name="EventType" value="ACTIVITY"/>
        <field name="EventName" formattingPattern="Metrics of {0} {1}">
            <field-locator locator="$.Topic" locator-type="Label"/>
            <field-locator locator="OperationType" locator-type="Activity"></field-locator>
        </field>

        <field name="ProducerSendMetrics" locator="$.Metrics" locator-type="Label">
            <parser-ref name="MetricsParser" aggregation="Merge"/>
        </field>

        <field name="clientID" locator="$.ClientId" locator-type="Label"/>
        <field name="Call" locator="$.CallName" locator-type="Label"/>

        <field name="Offset" locator="$.Offset.offset" locator-type="Label"/>
        <field name="Timestamp" locator="$.Offset.p_timestamp" locator-type="Label"/>
    </parser>

    <parser name="AckMetricsParser" class="com.jkoolcloud.tnt4j.streams.parsers.ActivityJsonParser" default-data-type="AsInput"
            manualFieldsOrder="true">
        <property name="ActivityDelim" value="EOF"/>

        <field name="topic" locator="$.Topic" locator-type="Label" required="true"/>
        <field name="OperationType" locator="$.Type" locator-type="Label" transparent="true">
            <field-map source="ProducerTopicMetrics" target="Producer"/>
            <field-map source="ConsumerTopicMetrics" target="Consumer"/>
        </field>

        <field name="EventType" value="ACTIVITY"/>
        <field name="EventName" formattingPattern="Metrics of {0} {1}">
            <field-locator locator="$.Topic" locator-type="Label"/>
            <field-locator locator="OperationType" locator-type="Activity"></field-locator>
        </field>

        <field name="ProducerAckMetrics" locator="$.Metrics" locator-type="Label">
            <parser-ref name="MetricsParser" aggregation="Merge"/>
        </field>

        <field name="LastOffset" locator="LastProduced" locator-type="Cache"/>
        <field name="LastOffsetTime" locator="LastProducedTime" locator-type="Cache"/>

        <field name="clientID" locator="$.ClientId" locator-type="Label"/>
        <field name="Call" locator="$.CallName" locator-type="Label"/>

        <field name="Offset" locator="$.Offset.offset" locator-type="Label"/>
        <field name="Timestamp" locator="$.Offset.p_timestamp" locator-type="Label"/>

        <field name="LatestPOffset" locator="Offset" locator-type="Activity" transparent="true">
            <field-transform lang="groovy" name="DiffTransform"><![CDATA[
                ${LastOffset} == null
                    ? $fieldValue
                    : Math.max(${LastOffset}, $fieldValue)
            ]]></field-transform>
        </field>
        <field name="LatestPTimestamp" locator="Timestamp" locator-type="Activity" transparent="true">
            <field-transform lang="groovy" name="DiffTransform"><![CDATA[
                ${LastOffset} == null
                    ? null
                    : (${Offset} >= ${LastOffset}
                        ? $fieldValue
                        : ${LastOffsetTime})
            ]]></field-transform>
        </field>
    </parser>

    <parser name="ConsumeMetricsParser" class="com.jkoolcloud.tnt4j.streams.parsers.ActivityJsonParser"
            default-data-type="AsInput" manualFieldsOrder="true">
        <property name="ActivityDelim" value="EOF"/>

        <field name="topic" locator="$.Topic" locator-type="Label" required="true"/>
        <field name="OperationType" locator="$.Type" locator-type="Label" transparent="true">
            <field-map source="ProducerTopicMetrics" target="Producer"/>
            <field-map source="ConsumerTopicMetrics" target="Consumer"/>
        </field>

        <field name="EventType" value="ACTIVITY"/>
        <field name="EventName" formattingPattern="Metrics of {0} {1}">
            <field-locator locator="$.Topic" locator-type="Label"/>
            <field-locator locator="OperationType" locator-type="Activity"></field-locator>
        </field>

        <field name="ConsumerConsumeMetrics" locator="$.Metrics" locator-type="Label">
            <parser-ref name="MetricsParser" aggregation="Merge"/>
        </field>

        <field name="LastOffset" locator="LastProduced" locator-type="Cache"/>
        <field name="LastOffsetTime" locator="LastProducedTime" locator-type="Cache"/>

        <field name="clientID" locator="$.ClientId" locator-type="Label"/>
        <field name="Call" locator="$.CallName" locator-type="Label"/>

        <field name="Offset" locator="$.Offset.offset" locator-type="Label"/>
        <field name="PTimestamp" locator="$.Offset.p_timestamp" locator-type="Label"/>
        <field name="PTimestampType" locator="$.Offset.p_timestamp_type" locator-type="Label"/>
        <field name="Timestamp" locator="$.Offset.c_timestamp" locator-type="Label"/>

        <field name="MessagesLag" locator="Offset" locator-type="Activity">
            <field-transform lang="groovy" name="DiffTransform"><![CDATA[
                ${LastOffset} != null && $fieldValue > 0
                    ? ${LastOffset} - $fieldValue
                    : 0
            ]]></field-transform>
        </field>

        <field name="MessagesLagTime" locator="Timestamp" locator-type="Activity">
            <field-transform lang="groovy" name="DiffTransform"><![CDATA[
                ${LastOffsetTime} != null && $fieldValue >= 0
                    ? $fieldValue - ${LastOffsetTime}
                    : 0
            ]]></field-transform>
        </field>

        <field name="CommitLagTime" locator="Timestamp" locator-type="Activity">
            <field-transform lang="groovy" name="DiffTransform"><![CDATA[
                $fieldValue >= 0
                    ? $fieldValue - ${PTimestamp}
                    : 0
			]]></field-transform>
        </field>

        <field name="AppendLagTime" locator="PTimestamp" locator-type="Activity">
            <field-transform lang="groovy" name="DiffTransform"><![CDATA[
                ${LastOffsetTime} != null && $fieldValue >= 0
                    ? $fieldValue - ${LastOffsetTime}
                    : 0
		    ]]></field-transform>
        </field>
    </parser>

    <parser name="CommitMetricsParser" class="com.jkoolcloud.tnt4j.streams.parsers.ActivityJsonParser" default-data-type="AsInput"
            manualFieldsOrder="true">
        <property name="ActivityDelim" value="EOF"/>

        <field name="topic" locator="$.Topic" locator-type="Label" required="true"/>
        <field name="OperationType" locator="$.Type" locator-type="Label" transparent="true">
            <field-map source="ProducerTopicMetrics" target="Producer"/>
            <field-map source="ConsumerTopicMetrics" target="Consumer"/>
        </field>

        <field name="EventType" value="ACTIVITY"/>
        <field name="EventName" formattingPattern="Metrics of {0} {1}">
            <field-locator locator="$.Topic" locator-type="Label"/>
            <field-locator locator="OperationType" locator-type="Activity"></field-locator>
        </field>

        <field name="ConsumerCommitMetrics" locator="$.Metrics" locator-type="Label">
            <parser-ref name="MetricsParser" aggregation="Merge"/>
        </field>

        <field name="clientID" locator="$.ClientId" locator-type="Label"/>
        <field name="Call" locator="$.CallName" locator-type="Label"/>

        <field name="Offset" locator="$.Offset.offset" locator-type="Label"/>
        <field name="Timestamp" locator="$.Offset.c_timestamp" locator-type="Label"/>
    </parser>

    <parser name="KafkaInterceptMetricsParser" class="com.jkoolcloud.tnt4j.streams.parsers.ActivityJsonParser" default-data-type="AsInput">
        <field name="message-text" locator="$.msg-text" locator-type="Label" transparent="true">
            <parser-ref name="AckMetricsParser" aggregation="Merge">
                <matchExp>JPATH:$[?(@.CallName == 'acknowledge' )]</matchExp>
            </parser-ref>
            <parser-ref name="SendMetricsParser" aggregation="Merge">
                <matchExp>JPATH:$[?(@.CallName == 'send' )]</matchExp>
            </parser-ref>
            <parser-ref name="CommitMetricsParser" aggregation="Merge">
                <matchExp>JPATH:$[?(@.CallName == 'commit' )]</matchExp>
            </parser-ref>
            <parser-ref name="ConsumeMetricsParser" aggregation="Merge">
                <matchExp>JPATH:$[?(@.CallName == 'consume' )]</matchExp>
            </parser-ref>
        </field>
    </parser>
    <!-- END of metrics parsing -->

    <!-- START of JMX metrics parsing -->

    <parser name="SnapshotParser" class="com.jkoolcloud.tnt4j.streams.parsers.ActivityMapParser" default-data-type="AsInput">
        <field name="EventName" locator="properties.ObjectName" locator-type="Label"/>
        <field name="EventType" value="SNAPSHOT"/>
        <field name="EElse" locator="properties.*" locator-type="Label"/>
    </parser>

    <parser name="TNTJSONParser" class="com.jkoolcloud.tnt4j.streams.parsers.ActivityJsonParser" default-data-type="AsInput">
        <property name="ActivityDelim" value="EOF"/>
        <field name="EventType" value="Activity"/>
        <field name="EventName" locator="$.operation" locator-type="Label"/>

        <embedded-activity name="Snapshots" locator="$.snapshots.[*]" locator-type="Label">
            <parser-ref name="SnapshotParser" aggregation="Relate"/>
        </embedded-activity>
    </parser>

    <parser name="KafkaMessageParser" class="com.jkoolcloud.tnt4j.streams.parsers.KafkaConsumerRecordParser">
        <field name="TNTEventData" locator="value" locator-type="Label" transparent="true">
            <parser-ref name="TNTJSONParser" aggregation="Merge">
                <matchExp>JPATH:$.snapshots</matchExp>
            </parser-ref>
        </field>
        <field name="KafkaInterceptMetricsData" locator="value" locator-type="Label" transparent="true">
            <parser-ref name="KafkaInterceptMetricsParser" aggregation="Merge">
                <matchExp>JPATH:$.msg-text</matchExp>
            </parser-ref>
        </field>
    </parser>

    <!--cache def -->

    <cache>
        <property name="MaxSize" value="100"/>
        <property name="ExpireDuration" value="55"/>
        <property name="Persisted" value="false"/>

        <entry id="LastProduced">
            <key>LastProduced-${topic}</key>
            <value>${LatestPOffset}</value>
        </entry>
        <entry id="LastProducedTime">
            <key>LastProducedTime-${topic}</key>
            <value>${LatestPTimestamp}</value>
        </entry>
    </cache>

    <!-- END of JMX metrics parsing -->

    <stream name="KafkaMetricsStream" class="com.jkoolcloud.tnt4j.streams.inputs.KafkaConsumerStream">
        <property name="HaltIfNoParser" value="false"/>
        <property name="RetryStateCheck" value="3"/>
        <property name="RetryInterval" value="5"/>
        <property name="Topic" value="tnt4j-kafka-interceptor-metrics"/><commit_msg>Added Correlator field.<commit_after><?xml version="1.0" encoding="utf-8"?>
<tnt-data-source
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="https://raw.githubusercontent.com/Nastel/tnt4j-streams/master/config/tnt-data-source.xsd">

    <!-- START of metrics parsing -->
    <parser name="MetricsParser" class="com.jkoolcloud.tnt4j.streams.parsers.ActivityMapParser">
        <field name="All" locator="*" locator-type="Label"/>
    </parser>

    <parser name="SendMetricsParser" class="com.jkoolcloud.tnt4j.streams.parsers.ActivityJsonParser" default-data-type="AsInput"
            manualFieldsOrder="true">
        <property name="ActivityDelim" value="EOF"/>

        <field name="topic" locator="$.Topic" locator-type="Label" required="true"/>
        <field name="OperationType" locator="$.Type" locator-type="Label" transparent="true">
            <field-map source="ProducerTopicMetrics" target="Producer"/>
            <field-map source="ConsumerTopicMetrics" target="Consumer"/>
        </field>

        <field name="EventType" value="ACTIVITY"/>
        <field name="EventName" formattingPattern="Metrics of {0} {1}">
            <field-locator locator="$.Topic" locator-type="Label"/>
            <field-locator locator="OperationType" locator-type="Activity"></field-locator>
        </field>

        <field name="ProducerSendMetrics" locator="$.Metrics" locator-type="Label">
            <parser-ref name="MetricsParser" aggregation="Merge"/>
        </field>

        <field name="clientID" locator="$.ClientId" locator-type="Label"/>
        <field name="Call" locator="$.CallName" locator-type="Label"/>

        <field name="Offset" locator="$.Offset.offset" locator-type="Label"/>
        <field name="Timestamp" locator="$.Offset.p_timestamp" locator-type="Label"/>

        <field name="Correlator" locator="$.Correlator" locator-type="Label"/>
    </parser>

    <parser name="AckMetricsParser" class="com.jkoolcloud.tnt4j.streams.parsers.ActivityJsonParser" default-data-type="AsInput"
            manualFieldsOrder="true">
        <property name="ActivityDelim" value="EOF"/>

        <field name="topic" locator="$.Topic" locator-type="Label" required="true"/>
        <field name="OperationType" locator="$.Type" locator-type="Label" transparent="true">
            <field-map source="ProducerTopicMetrics" target="Producer"/>
            <field-map source="ConsumerTopicMetrics" target="Consumer"/>
        </field>

        <field name="EventType" value="ACTIVITY"/>
        <field name="EventName" formattingPattern="Metrics of {0} {1}">
            <field-locator locator="$.Topic" locator-type="Label"/>
            <field-locator locator="OperationType" locator-type="Activity"></field-locator>
        </field>

        <field name="ProducerAckMetrics" locator="$.Metrics" locator-type="Label">
            <parser-ref name="MetricsParser" aggregation="Merge"/>
        </field>

        <field name="LastOffset" locator="LastProduced" locator-type="Cache"/>
        <field name="LastOffsetTime" locator="LastProducedTime" locator-type="Cache"/>

        <field name="clientID" locator="$.ClientId" locator-type="Label"/>
        <field name="Call" locator="$.CallName" locator-type="Label"/>

        <field name="Offset" locator="$.Offset.offset" locator-type="Label"/>
        <field name="Timestamp" locator="$.Offset.p_timestamp" locator-type="Label"/>

        <field name="LatestPOffset" locator="Offset" locator-type="Activity" transparent="true">
            <field-transform lang="groovy" name="DiffTransform"><![CDATA[
                ${LastOffset} == null
                    ? $fieldValue
                    : Math.max(${LastOffset}, $fieldValue)
            ]]></field-transform>
        </field>
        <field name="LatestPTimestamp" locator="Timestamp" locator-type="Activity" transparent="true">
            <field-transform lang="groovy" name="DiffTransform"><![CDATA[
                ${LastOffset} == null
                    ? null
                    : (${Offset} >= ${LastOffset}
                        ? $fieldValue
                        : ${LastOffsetTime})
            ]]></field-transform>
        </field>

        <field name="Correlator" locator="$.Correlator" locator-type="Label"/>
    </parser>

    <parser name="ConsumeMetricsParser" class="com.jkoolcloud.tnt4j.streams.parsers.ActivityJsonParser"
            default-data-type="AsInput" manualFieldsOrder="true">
        <property name="ActivityDelim" value="EOF"/>

        <field name="topic" locator="$.Topic" locator-type="Label" required="true"/>
        <field name="OperationType" locator="$.Type" locator-type="Label" transparent="true">
            <field-map source="ProducerTopicMetrics" target="Producer"/>
            <field-map source="ConsumerTopicMetrics" target="Consumer"/>
        </field>

        <field name="EventType" value="ACTIVITY"/>
        <field name="EventName" formattingPattern="Metrics of {0} {1}">
            <field-locator locator="$.Topic" locator-type="Label"/>
            <field-locator locator="OperationType" locator-type="Activity"></field-locator>
        </field>

        <field name="ConsumerConsumeMetrics" locator="$.Metrics" locator-type="Label">
            <parser-ref name="MetricsParser" aggregation="Merge"/>
        </field>

        <field name="LastOffset" locator="LastProduced" locator-type="Cache"/>
        <field name="LastOffsetTime" locator="LastProducedTime" locator-type="Cache"/>

        <field name="clientID" locator="$.ClientId" locator-type="Label"/>
        <field name="Call" locator="$.CallName" locator-type="Label"/>

        <field name="Offset" locator="$.Offset.offset" locator-type="Label"/>
        <field name="PTimestamp" locator="$.Offset.p_timestamp" locator-type="Label"/>
        <field name="PTimestampType" locator="$.Offset.p_timestamp_type" locator-type="Label"/>
        <field name="Timestamp" locator="$.Offset.c_timestamp" locator-type="Label"/>

        <field name="MessagesLag" locator="Offset" locator-type="Activity">
            <field-transform lang="groovy" name="DiffTransform"><![CDATA[
                ${LastOffset} != null && $fieldValue > 0
                    ? ${LastOffset} - $fieldValue
                    : 0
            ]]></field-transform>
        </field>

        <field name="MessagesLagTime" locator="Timestamp" locator-type="Activity">
            <field-transform lang="groovy" name="DiffTransform"><![CDATA[
                ${LastOffsetTime} != null && $fieldValue >= 0
                    ? $fieldValue - ${LastOffsetTime}
                    : 0
            ]]></field-transform>
        </field>

        <field name="CommitLagTime" locator="Timestamp" locator-type="Activity">
            <field-transform lang="groovy" name="DiffTransform"><![CDATA[
                $fieldValue >= 0
                    ? $fieldValue - ${PTimestamp}
                    : 0
			]]></field-transform>
        </field>

        <field name="AppendLagTime" locator="PTimestamp" locator-type="Activity">
            <field-transform lang="groovy" name="DiffTransform"><![CDATA[
                ${LastOffsetTime} != null && $fieldValue >= 0
                    ? $fieldValue - ${LastOffsetTime}
                    : 0
		    ]]></field-transform>
        </field>

        <field name="Correlator" locator="$.Correlator" locator-type="Label"/>
    </parser>

    <parser name="CommitMetricsParser" class="com.jkoolcloud.tnt4j.streams.parsers.ActivityJsonParser" default-data-type="AsInput"
            manualFieldsOrder="true">
        <property name="ActivityDelim" value="EOF"/>

        <field name="topic" locator="$.Topic" locator-type="Label" required="true"/>
        <field name="OperationType" locator="$.Type" locator-type="Label" transparent="true">
            <field-map source="ProducerTopicMetrics" target="Producer"/>
            <field-map source="ConsumerTopicMetrics" target="Consumer"/>
        </field>

        <field name="EventType" value="ACTIVITY"/>
        <field name="EventName" formattingPattern="Metrics of {0} {1}">
            <field-locator locator="$.Topic" locator-type="Label"/>
            <field-locator locator="OperationType" locator-type="Activity"></field-locator>
        </field>

        <field name="ConsumerCommitMetrics" locator="$.Metrics" locator-type="Label">
            <parser-ref name="MetricsParser" aggregation="Merge"/>
        </field>

        <field name="clientID" locator="$.ClientId" locator-type="Label"/>
        <field name="Call" locator="$.CallName" locator-type="Label"/>

        <field name="Offset" locator="$.Offset.offset" locator-type="Label"/>
        <field name="Timestamp" locator="$.Offset.c_timestamp" locator-type="Label"/>

        <field name="Correlator" locator="$.Correlator" locator-type="Label"/>
    </parser>

    <parser name="KafkaInterceptMetricsParser" class="com.jkoolcloud.tnt4j.streams.parsers.ActivityJsonParser" default-data-type="AsInput">
        <field name="message-text" locator="$.msg-text" locator-type="Label" transparent="true">
            <parser-ref name="AckMetricsParser" aggregation="Merge">
                <matchExp>JPATH:$[?(@.CallName == 'acknowledge' )]</matchExp>
            </parser-ref>
            <parser-ref name="SendMetricsParser" aggregation="Merge">
                <matchExp>JPATH:$[?(@.CallName == 'send' )]</matchExp>
            </parser-ref>
            <parser-ref name="CommitMetricsParser" aggregation="Merge">
                <matchExp>JPATH:$[?(@.CallName == 'commit' )]</matchExp>
            </parser-ref>
            <parser-ref name="ConsumeMetricsParser" aggregation="Merge">
                <matchExp>JPATH:$[?(@.CallName == 'consume' )]</matchExp>
            </parser-ref>
        </field>
    </parser>
    <!-- END of metrics parsing -->

    <!-- START of JMX metrics parsing -->
    <parser name="SnapshotParser" class="com.jkoolcloud.tnt4j.streams.parsers.ActivityMapParser" default-data-type="AsInput">
        <field name="EventName" locator="properties.ObjectName" locator-type="Label"/>
        <field name="EventType" value="SNAPSHOT"/>
        <field name="JMXMetrics" locator="properties.*" locator-type="Label"/>
        <field name="Category" value="Kafka JMX metrics snapshot"/>
    </parser>

    <parser name="TNTJSONParser" class="com.jkoolcloud.tnt4j.streams.parsers.ActivityJsonParser" default-data-type="AsInput">
        <property name="ActivityDelim" value="EOF"/>
        <field name="EventType" value="Activity"/>
        <field name="EventName" locator="$.operation" locator-type="Label"/>
        <field name="Correlator" locator="$.corrid" locator-type="Label"/>

        <embedded-activity name="Snapshots" locator="$.snapshots.[*]" locator-type="Label">
            <parser-ref name="SnapshotParser" aggregation="Relate"/>
        </embedded-activity>
    </parser>

    <parser name="KafkaMessageParser" class="com.jkoolcloud.tnt4j.streams.parsers.KafkaConsumerRecordParser">
        <field name="TNTEventData" locator="value" locator-type="Label" transparent="true">
            <parser-ref name="TNTJSONParser" aggregation="Merge">
                <matchExp>JPATH:$.snapshots</matchExp>
            </parser-ref>
        </field>
        <field name="KafkaInterceptMetricsData" locator="value" locator-type="Label" transparent="true">
            <parser-ref name="KafkaInterceptMetricsParser" aggregation="Merge">
                <matchExp>JPATH:$.msg-text</matchExp>
            </parser-ref>
        </field>
    </parser>
    <!-- END of JMX metrics parsing -->

    <!--cache def -->
    <cache>
        <property name="MaxSize" value="100"/>
        <property name="ExpireDuration" value="55"/>
        <property name="Persisted" value="false"/>

        <entry id="LastProduced">
            <key>LastProduced-${topic}</key>
            <value>${LatestPOffset}</value>
        </entry>
        <entry id="LastProducedTime">
            <key>LastProducedTime-${topic}</key>
            <value>${LatestPTimestamp}</value>
        </entry>
    </cache>

    <stream name="KafkaMetricsStream" class="com.jkoolcloud.tnt4j.streams.inputs.KafkaConsumerStream">
        <property name="HaltIfNoParser" value="false"/>
        <property name="RetryStateCheck" value="3"/>
        <property name="RetryInterval" value="5"/>
        <property name="Topic" value="tnt4j-kafka-interceptor-metrics"/>
{'fraction_non_alphanumeric': 0.16531730454694143, 'fraction_numerical': 0.003516519463526333, 'mean_word_length': 2.960006476683938, 'pattern_counts': {'":': 0, '<': 407, '<?xml version=': 2, '>': 417, 'https://': 2, 'lorem ipsum': 0, 'www.': 2, 'xml': 4}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6646818', 'n_tokens_mistral': 7150, 'n_tokens_neox': 6885, 'n_words': 1139}