Integrate inline catalog and json validator with mqtt binding

incubator/command-generate/src/main/java/io/aklivity/zilla/runtime/command/generate/internal/asyncapi/mqtt/proxy/AsyncApiMqttProxyConfigGenerator.java

@@ -293,7 +293,7 @@ private <C> CatalogedConfigBuilder<C> injectJsonSchemas(
             {
                 cataloged
                     .schema()
-                        .schema(schema)
+                        .subject(schema)
                         .build()
                     .build();
             }

incubator/command-generate/src/test/resources/io/aklivity/zilla/runtime/command/generate/internal/asyncapi/mqtt/proxy/complete/zilla.yaml

@@ -37,7 +37,7 @@ bindings:
           type: json
           catalog:
             catalog0:
-            - schema: items
+            - subject: items
     routes:
     - exit: mqtt_client0
       when:

incubator/command-generate/src/test/resources/io/aklivity/zilla/runtime/command/generate/internal/asyncapi/mqtt/proxy/validator/zilla.yaml

@@ -20,8 +20,8 @@ bindings:
           type: json
           catalog:
             catalog0:
-            - schema: items
-            - schema: things
+            - subject: items
+            - subject: things
     routes:
     - exit: mqtt_client0
       when:

incubator/validator-json/src/main/java/io/aklivity/zilla/runtime/validator/json/JsonValidator.java

@@ -104,7 +104,7 @@ private boolean validate(
         else if (catalog != null)
         {
             schemaId = handler.resolve(catalog.subject, catalog.version);
-            if (schemaId > 0)
+            if (schemaId != 0)
             {
                 schema = handler.resolve(schemaId);
             }

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

@@ -945,13 +945,15 @@ private int decodePublish(
                     reasonCode = PAYLOAD_FORMAT_INVALID;
                     server.onDecodeError(traceId, authorization, reasonCode);
                     server.decoder = decodeIgnoreAll;
+                    break decode;
                 }
 
                 if (mqttPublishHeaderRO.payloadFormat.equals(MqttPayloadFormat.TEXT) && invalidUtf8(payload))
                 {
                     reasonCode = PAYLOAD_FORMAT_INVALID;
                     server.onDecodeError(traceId, authorization, reasonCode);
                     server.decoder = decodeIgnoreAll;
+                    break decode;
                 }
 
                 boolean canPublish = MqttState.initialOpened(publisher.state);