feat: replace static event name with dynamic based on event id

incubator/catalog-filesystem.spec/src/main/scripts/io/aklivity/zilla/specs/catalog/filesystem/config/event.yaml

@@ -23,7 +23,8 @@ telemetry:
         events:
           - qname: test.catalog0
             id: catalog.filesystem.file.not.found
-            message: FILE_NOT_FOUND asyncapi/kafka.yaml
+            name: CATALOG_FILESYSTEM_FILE_NOT_FOUND
+            message: Unable to find file at (asyncapi/kafka.yaml) on the host filesystem.
 catalogs:
   catalog0:
     type: filesystem

incubator/catalog-filesystem/src/main/java/io/aklivity/zilla/runtime/catalog/filesystem/internal/FilesystemEventFormatter.java

@@ -25,10 +25,8 @@
 
 public final class FilesystemEventFormatter implements EventFormatterSpi
 {
-    private static final String FILE_NOT_FOUND = "FILE_NOT_FOUND %s";
-
     private final EventFW eventRO = new EventFW();
-    private final FilesystemEventExFW schemaRegistryEventExRO = new FilesystemEventExFW();
+    private final FilesystemEventExFW filesystemEventExRO = new FilesystemEventExFW();
 
     FilesystemEventFormatter(
         Configuration config)
@@ -41,15 +39,15 @@ public String format(
         int length)
     {
         final EventFW event = eventRO.wrap(buffer, index, index + length);
-        final FilesystemEventExFW extension = schemaRegistryEventExRO
+        final FilesystemEventExFW extension = filesystemEventExRO
             .wrap(event.extension().buffer(), event.extension().offset(), event.extension().limit());
         String result = null;
         switch (extension.kind())
         {
         case FILE_NOT_FOUND:
         {
             FilesystemFileNotFoundExFW ex = extension.fileNotFound();
-            result = String.format(FILE_NOT_FOUND, asString(ex.location()));
+            result = String.format("Unable to find file at (%s) on the host filesystem.", asString(ex.location()));
             break;
         }
         }

runtime/binding-echo/src/test/java/io/aklivity/zilla/runtime/binding/echo/internal/bench/EchoWorker.java

@@ -263,6 +263,13 @@ public int supplyEventId(
         return 0;
     }
 
+    @Override
+    public String supplyEventName(
+        int eventId)
+    {
+        return "";
+    }
+
     @Override
     public BindingHandler streamFactory()
     {

runtime/binding-http/src/main/java/io/aklivity/zilla/runtime/binding/http/internal/HttpEventFormatter.java

@@ -26,8 +26,6 @@
 
 public final class HttpEventFormatter implements EventFormatterSpi
 {
-    private static final String REQUEST_ACCEPTED_FORMAT = "REQUEST_ACCEPTED %s %s %s %s %s";
-
     private final EventFW eventRO = new EventFW();
     private final HttpEventExFW httpEventExRO = new HttpEventExFW();
 
@@ -50,7 +48,7 @@ public String format(
         case REQUEST_ACCEPTED:
         {
             HttpRequestAcceptedExFW ex = extension.requestAccepted();
-            result = String.format(REQUEST_ACCEPTED_FORMAT, identity(ex.identity()), asString(ex.scheme()), asString(ex.method()),
+            result = String.format("%s %s %s://%s%s", identity(ex.identity()), asString(ex.method()), asString(ex.scheme()),
                 asString(ex.authority()), asString(ex.path()));
             break;
         }

runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/KafkaEventFormatter.java

@@ -27,9 +27,6 @@
 
 public final class KafkaEventFormatter implements EventFormatterSpi
 {
-    private static final String AUTHORIZATION_FAILED_FORMAT = "AUTHORIZATION_FAILED %s";
-    private static final String API_VERSION_REJECTED_FORMAT = "API_VERSION_REJECTED %d %d";
-
     private final EventFW eventRO = new EventFW();
     private final KafkaEventExFW kafkaEventExRO = new KafkaEventExFW();
 
@@ -52,22 +49,22 @@ public String format(
         case AUTHORIZATION_FAILED:
         {
             KafkaAuthorizationFailedExFW ex = extension.authorizationFailed();
-            result = String.format(AUTHORIZATION_FAILED_FORMAT, identity(ex.identity()));
+            result = String.format("Unable to authenticate client with identity (%s).", asString(ex.identity()));
             break;
         }
         case API_VERSION_REJECTED:
         {
             final KafkaApiVersionRejectedExFW ex = extension.apiVersionRejected();
-            result = String.format(API_VERSION_REJECTED_FORMAT, ex.apiKey(), ex.apiVersion());
+            result = String.format("%d %d", ex.apiKey(), ex.apiVersion());
         }
         }
         return result;
     }
 
-    private static String identity(
-        StringFW identity)
+    private static String asString(
+            StringFW stringFW)
     {
-        int length = identity.length();
-        return length <= 0 ? "-" : identity.asString();
+        String s = stringFW.asString();
+        return s == null ? "" : s;
     }
 }

runtime/binding-mqtt-kafka/src/main/java/io/aklivity/zilla/runtime/binding/mqtt/kafka/internal/MqttKafkaEventContext.java

@@ -53,13 +53,13 @@ public MqttKafkaEventContext(
     public void onMqttConnectionReset(
         long traceId,
         long bindingId,
-        String16FW reason)
+        String16FW topic)
     {
         MqttKafkaEventExFW extension = mqttKafkaEventExRW
             .wrap(extensionBuffer, 0, extensionBuffer.capacity())
             .nonCompactSessionsTopic(e -> e
                 .typeId(NON_COMPACT_SESSIONS_TOPIC.value())
-                .reason(reason))
+                .topic(topic))
             .build();
         EventFW event = eventRW
             .wrap(eventBuffer, 0, eventBuffer.capacity())

runtime/binding-mqtt-kafka/src/main/java/io/aklivity/zilla/runtime/binding/mqtt/kafka/internal/MqttKafkaEventFormatter.java

@@ -19,14 +19,12 @@
 import io.aklivity.zilla.runtime.binding.mqtt.kafka.internal.types.String16FW;
 import io.aklivity.zilla.runtime.binding.mqtt.kafka.internal.types.event.EventFW;
 import io.aklivity.zilla.runtime.binding.mqtt.kafka.internal.types.event.MqttKafkaEventExFW;
-import io.aklivity.zilla.runtime.binding.mqtt.kafka.internal.types.event.MqttKafkaResetMqttConnectionExFW;
+import io.aklivity.zilla.runtime.binding.mqtt.kafka.internal.types.event.MqttKafkaNonCompactSessionsTopicExFW;
 import io.aklivity.zilla.runtime.engine.Configuration;
 import io.aklivity.zilla.runtime.engine.event.EventFormatterSpi;
 
 public final class MqttKafkaEventFormatter implements EventFormatterSpi
 {
-    private static final String NON_COMPACT_SESSIONS_TOPIC_FORMAT = "NON COMPACT SESSIONS TOPIC - %s";
-
     private final EventFW eventRO = new EventFW();
     private final MqttKafkaEventExFW mqttKafkaEventExRO = new MqttKafkaEventExFW();
 
@@ -48,8 +46,11 @@ public String format(
         {
         case NON_COMPACT_SESSIONS_TOPIC:
         {
-            MqttKafkaResetMqttConnectionExFW ex = extension.nonCompactSessionsTopic();
-            result = String.format(NON_COMPACT_SESSIONS_TOPIC_FORMAT, asString(ex.reason()));
+            MqttKafkaNonCompactSessionsTopicExFW ex = extension.nonCompactSessionsTopic();
+            result = String.format(
+                    "The sessions topic (%s) is not log compacted. Update the cleanup policy to enable log compaction.",
+                    asString(ex.topic())
+            );
             break;
         }
         }

runtime/binding-mqtt-kafka/src/main/java/io/aklivity/zilla/runtime/binding/mqtt/kafka/internal/stream/MqttKafkaSessionFactory.java

@@ -170,7 +170,6 @@ public class MqttKafkaSessionFactory implements MqttKafkaStreamFactory
     public static final int MQTT_NOT_AUTHORIZED = 0x87;
     public static final int MQTT_IMPLEMENTATION_SPECIFIC_ERROR = 0x83;
     public static final String MQTT_INVALID_SESSION_TIMEOUT_REASON = "Invalid session expiry interval";
-    public static final String16FW MQTT_NON_COMPACT_SESSIONS_TOPIC = new String16FW("Sessions Kafka topic in non-compacted");
     private static final KafkaConfigFW CONFIG_COMPACT_CLEANUP_POLICY = new KafkaConfigFW.Builder()
         .wrap(new UnsafeBuffer(new byte[25]), 0, 25)
         .name("cleanup.policy")
@@ -3469,7 +3468,7 @@ protected void onKafkaBegin(
                             .build();
                         delegate.doMqttWindow(authorization, traceId, 0, 0, 0);
                         delegate.doMqttReset(traceId, mqttResetEx);
-                        events.onMqttConnectionReset(traceId, routedId, MQTT_NON_COMPACT_SESSIONS_TOPIC);
+                        events.onMqttConnectionReset(traceId, routedId, delegate.sessionsTopic);
                         doKafkaWindow(traceId, authorization, 0, 0, 0, 0, 0);
                         doKafkaAbort(traceId, authorization);
                         break onKafkaBegin;

runtime/binding-mqtt/src/main/java/io/aklivity/zilla/runtime/binding/mqtt/internal/MqttEventFormatter.java

@@ -26,8 +26,6 @@
 
 public final class MqttEventFormatter implements EventFormatterSpi
 {
-    private static final String CLIENT_CONNECTED_FORMAT = "CLIENT_CONNECTED %s %s";
-
     private final EventFW eventRO = new EventFW();
     private final MqttEventExFW mqttEventExRO = new MqttEventExFW();
 
@@ -50,7 +48,10 @@ public String format(
         case CLIENT_CONNECTED:
         {
             MqttClientConnectedExFW ex = extension.clientConnected();
-            result = String.format(CLIENT_CONNECTED_FORMAT, identity(ex.identity()), asString(ex.clientId()));
+            result = String.format("Session authorization (%s) was successful for client id (%s).",
+                    identity(ex.identity()),
+                    asString(ex.clientId())
+            );
             break;
         }
         }

runtime/binding-tcp/src/main/java/io/aklivity/zilla/runtime/binding/tcp/internal/TcpEventFormatter.java

@@ -26,8 +26,6 @@
 
 public final class TcpEventFormatter implements EventFormatterSpi
 {
-    private static final String DNS_FAILED_FORMAT = "DNS_FAILED %s";
-
     private final EventFW eventRO = new EventFW();
     private final TcpEventExFW tcpEventExRO = new TcpEventExFW();
 
@@ -50,7 +48,7 @@ public String format(
         case DNS_FAILED:
         {
             final TcpDnsFailedExFW ex = extension.dnsFailed();
-            result = String.format(DNS_FAILED_FORMAT, asString(ex.address()));
+            result = String.format("Unable to resolve host dns for address (%s).", asString(ex.address()));
             break;
         }
         }

runtime/binding-tls/src/main/java/io/aklivity/zilla/runtime/binding/tls/internal/TlsEventFormatter.java

@@ -24,12 +24,6 @@
 
 public final class TlsEventFormatter implements EventFormatterSpi
 {
-    private static final String TLS_FAILED_FORMAT = "TLS_FAILED";
-    private static final String PROTOCOL_REJECTED_FORMAT = "PROTOCOL_REJECTED";
-    private static final String KEY_REJECTED_FORMAT = "KEY_REJECTED";
-    private static final String PEER_NOT_VERIFIED_FORMAT = "PEER_NOT_VERIFIED";
-    private static final String HANDSHAKE_FAILED_FORMAT = "HANDSHAKE_FAILED";
-
     private final EventFW eventRO = new EventFW();
     private final TlsEventExFW tlsEventExRO = new TlsEventExFW();
 
@@ -51,27 +45,27 @@ public String format(
         {
         case TLS_FAILED:
         {
-            result = TLS_FAILED_FORMAT;
+            result = "There was a generic error detected by an SSL subsystem.";
             break;
         }
         case TLS_PROTOCOL_REJECTED:
         {
-            result = PROTOCOL_REJECTED_FORMAT;
+            result = "There was an error in the operation of the SSL protocol.";
             break;
         }
         case TLS_KEY_REJECTED:
         {
-            result = KEY_REJECTED_FORMAT;
+            result = "Bad SSL key due to misconfiguration of the server or client SSL certificate and private key.";
             break;
         }
         case TLS_PEER_NOT_VERIFIED:
         {
-            result = PEER_NOT_VERIFIED_FORMAT;
+            result = "The peer's identity could not be verified.";
             break;
         }
         case TLS_HANDSHAKE_FAILED:
         {
-            result = HANDSHAKE_FAILED_FORMAT;
+            result = "The client and server could not negotiate the desired level of security.";
             break;
         }
         }

runtime/binding-tls/src/test/java/io/aklivity/zilla/runtime/binding/tls/internal/bench/TlsWorker.java

@@ -331,6 +331,13 @@ public int supplyEventId(
         return 0;
     }
 
+    @Override
+    public String supplyEventName(
+        int eventId)
+    {
+        return "";
+    }
+
     @Override
     public BindingHandler streamFactory()
     {

runtime/catalog-apicurio/src/main/java/io/aklivity/zilla/runtime/catalog/apicurio/internal/ApicurioEventContext.java

@@ -44,7 +44,7 @@ public class ApicurioEventContext
     private final int staleArtifactID;
     private final int unretrievableArtifactId;
     private final int retrievableArtifactSubjectVersionId;
-    private final int retrievableArtifactId;
+    private final int retrievedArtifactId;
     private final MessageConsumer eventWriter;
     private final Clock clock;
 
@@ -57,8 +57,8 @@ public ApicurioEventContext(
         this.staleArtifactID = context.supplyEventId("catalog.apicurio.unretrievable.artifact.subject.version.stale.artifact");
         this.unretrievableArtifactId = context.supplyEventId("catalog.apicurio.unretrievable.artifact.id");
         this.retrievableArtifactSubjectVersionId = context.supplyEventId(
-            "catalog.apicurio.retrievable.artifact.subject.version");
-        this.retrievableArtifactId = context.supplyEventId("catalog.apicurio.retrievable.artifact.id");
+            "catalog.apicurio.retrieved.artifact.subject.version");
+        this.retrievedArtifactId = context.supplyEventId("catalog.apicurio.retrieved.artifact.id");
         this.eventWriter = context.supplyEventWriter();
         this.clock = context.clock();
     }
@@ -142,7 +142,7 @@ public void onRetrievableArtifactSubjectVersion(
     {
         ApicurioEventExFW extension = apicurioEventExRW
             .wrap(extensionBuffer, 0, extensionBuffer.capacity())
-            .retrievableArtifactSubjectVersion(e -> e
+            .retrievedArtifactSubjectVersion(e -> e
                 .typeId(RETRIEVED_ARTIFACT_SUBJECT_VERSION.value())
                 .subject(subject)
                 .version(version)
@@ -165,14 +165,14 @@ public void onRetrievableArtifactId(
     {
         ApicurioEventExFW extension = apicurioEventExRW
             .wrap(extensionBuffer, 0, extensionBuffer.capacity())
-            .retrievableArtifactId(e -> e
+            .retrievedArtifactId(e -> e
                 .typeId(RETRIEVED_ARTIFACT_ID.value())
                 .artifactId(artifactId)
             )
             .build();
         EventFW event = eventRW
             .wrap(eventBuffer, 0, eventBuffer.capacity())
-            .id(retrievableArtifactId)
+            .id(retrievedArtifactId)
             .timestamp(clock.millis())
             .traceId(0L)
             .namespacedId(catalogId)

runtime/catalog-apicurio/src/main/java/io/aklivity/zilla/runtime/catalog/apicurio/internal/ApicurioEventFormatter.java

@@ -18,8 +18,8 @@
 
 import io.aklivity.zilla.runtime.catalog.apicurio.internal.types.StringFW;
 import io.aklivity.zilla.runtime.catalog.apicurio.internal.types.event.ApicurioEventExFW;
-import io.aklivity.zilla.runtime.catalog.apicurio.internal.types.event.ApicurioRetrievableArtifactIdExFW;
-import io.aklivity.zilla.runtime.catalog.apicurio.internal.types.event.ApicurioRetrievableArtifactSubjectVersionExFW;
+import io.aklivity.zilla.runtime.catalog.apicurio.internal.types.event.ApicurioRetrievedArtifactIdExFW;
+import io.aklivity.zilla.runtime.catalog.apicurio.internal.types.event.ApicurioRetrievedArtifactSubjectVersionExFW;
 import io.aklivity.zilla.runtime.catalog.apicurio.internal.types.event.ApicurioUnretrievableArtifactIdExFW;
 import io.aklivity.zilla.runtime.catalog.apicurio.internal.types.event.ApicurioUnretrievableArtifactSubjectVersionExFW;
 import io.aklivity.zilla.runtime.catalog.apicurio.internal.types.event.ApicurioUnretrievableArtifactSubjectVersionStaleArtifactExFW;
@@ -29,13 +29,6 @@
 
 public final class ApicurioEventFormatter implements EventFormatterSpi
 {
-    private static final String UNRETRIEVABLE_ARTIFACT_SUBJECT_VERSION = "UNRETRIEVABLE_ARTIFACT %s %s";
-    private static final String UNRETRIEVABLE_ARTIFACT_SUBJECT_VERSION_STALE_ARTIFACT =
-        "UNRETRIEVABLE_ARTIFACT %s %s, USING_STALE_ARTIFACT %d";
-    private static final String UNRETRIEVABLE_ARTIFACT_ID = "UNRETRIEVABLE_ARTIFACT_ID %d";
-    private static final String RETRIEVED_ARTIFACT_SUBJECT_VERSION = "RETRIEVED_ARTIFACT_SUBJECT_VERSION %s %s";
-    private static final String RETRIEVED_ARTIFACT_ID = "RETRIEVED_ARTIFACT_ID %d";
-
     private final EventFW eventRO = new EventFW();
     private final ApicurioEventExFW schemaRegistryEventExRO = new ApicurioEventExFW();
 
@@ -58,33 +51,44 @@ public String format(
         case UNRETRIEVABLE_ARTIFACT_SUBJECT_VERSION:
         {
             ApicurioUnretrievableArtifactSubjectVersionExFW ex = extension.unretrievableArtifactSubjectVersion();
-            result = String.format(UNRETRIEVABLE_ARTIFACT_SUBJECT_VERSION, asString(ex.subject()), asString(ex.version()));
+            result = String.format(
+                    "Unable to fetch artifact for subject %s with version %s.",
+                    asString(ex.subject()),
+                    asString(ex.version())
+            );
             break;
         }
         case UNRETRIEVABLE_ARTIFACT_SUBJECT_VERSION_STALE_ARTIFACT:
         {
             ApicurioUnretrievableArtifactSubjectVersionStaleArtifactExFW ex = extension
                 .unretrievableArtifactSubjectVersionStaleArtifact();
-            result = String.format(UNRETRIEVABLE_ARTIFACT_SUBJECT_VERSION_STALE_ARTIFACT, asString(ex.subject()),
-                asString(ex.version()), ex.artifactId());
+            result = String.format(
+                    "Unable to fetch artifact for subject %s with version %s; using stale artifact with id %d.",
+                    asString(ex.subject()),
+                    asString(ex.version()),
+                    ex.artifactId()
+            );
             break;
         }
         case UNRETRIEVABLE_ARTIFACT_ID:
         {
             ApicurioUnretrievableArtifactIdExFW ex = extension.unretrievableArtifactId();
-            result = String.format(UNRETRIEVABLE_ARTIFACT_ID, ex.artifactId());
+            result = String.format("Unable to fetch artifact id %d.", ex.artifactId());
             break;
         }
         case RETRIEVED_ARTIFACT_SUBJECT_VERSION:
         {
-            ApicurioRetrievableArtifactSubjectVersionExFW ex = extension.retrievableArtifactSubjectVersion();
-            result = String.format(RETRIEVED_ARTIFACT_SUBJECT_VERSION, asString(ex.subject()), asString(ex.version()));
+            ApicurioRetrievedArtifactSubjectVersionExFW ex = extension.retrievedArtifactSubjectVersion();
+            result = String.format("Successfully fetched artifact for subject %s with version %s.",
+                    asString(ex.subject()),
+                    asString(ex.version())
+            );
             break;
         }
         case RETRIEVED_ARTIFACT_ID:
         {
-            ApicurioRetrievableArtifactIdExFW ex = extension.retrievableArtifactId();
-            result = String.format(RETRIEVED_ARTIFACT_ID, ex.artifactId());
+            ApicurioRetrievedArtifactIdExFW ex = extension.retrievedArtifactId();
+            result = String.format("Successfully fetched artifact id %d.", ex.artifactId());
             break;
         }
         }