Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Schema fixes + avoiding duplicate reply begin on mqtt-kafka subscribe stream #872

Merged
merged 11 commits into from
Mar 25, 2024
Merged

Schema fixes + avoiding duplicate reply begin on mqtt-kafka subscribe stream #872

Show file tree
Hide file tree
Changes from 10 commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
56dce16
Fix qvault on asyncapi composite binding
bmaidics Mar 17, 2024
abd1c3f
Review
bmaidics Mar 18, 2024
4ba7ec8
remove pcap
bmaidics Mar 18, 2024
90a6ab8
Reference options in TLS
bmaidics Mar 19, 2024
9855486
Fix path
bmaidics Mar 19, 2024
c808627
commit
bmaidics Mar 20, 2024
7ccffe8
fixes
bmaidics Mar 20, 2024
9c61d86
Merge remote-tracking branch 'upstream/develop' into schema_fixes
bmaidics Mar 20, 2024
dc8decb
Fix tests
bmaidics Mar 25, 2024
8c098c9
Fix dependency
bmaidics Mar 25, 2024
23c1e9b
Revert mqtt-kafka schema changes
bmaidics Mar 25, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions incubator/binding-asyncapi.spec/NOTICE
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,10 @@ This project includes:
zilla::specs::binding-mqtt-kafka.spec under Aklivity Community License Agreement
zilla::specs::binding-mqtt.spec under The Apache Software License, Version 2.0
zilla::specs::binding-proxy.spec under The Apache Software License, Version 2.0
zilla::specs::binding-tcp.spec under The Apache Software License, Version 2.0
zilla::specs::binding-tls.spec under The Apache Software License, Version 2.0
zilla::specs::engine.spec under The Apache Software License, Version 2.0
zilla::specs::vault-filesystem.spec under The Apache Software License, Version 2.0


This project also includes code under copyright of the following entities:
Expand Down
10 changes: 10 additions & 0 deletions incubator/binding-asyncapi.spec/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,16 @@
<artifactId>binding-mqtt-kafka.spec</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>binding-tcp.spec</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>binding-tls.spec</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
Expand Down
43 changes: 17 additions & 26 deletions ...c/main/scripts/io/aklivity/zilla/specs/binding/asyncapi/schema/asyncapi.schema.patch.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,15 +52,24 @@
"maxProperties": 1
}
},
"tcp": "#/$defs/binding/tcp/options",
"tls": "#/$defs/binding/tls/options",
"tcp":
{
"$ref": "#/$defs/options/binding/tcp"
},
"tls":
{
"$ref": "#/$defs/options/binding/tls"
},
"http":
{
"title": "Http",
"type": "object",
"properties":
{
"authorization": "$defs/options/binding/http/authorization"
"authorization":
{
"$ref": "#/$defs/options/binding/http/authorization"
}
},
"additionalProperties": false
},
Expand All @@ -70,7 +79,10 @@
"type": "object",
"properties":
{
"sasl": "$defs/options/binding/kafka/sasl"
"sasl":
{
"$ref": "#/$defs/options/binding/kafka/sasl"
}
},
"additionalProperties": false
},
Expand All @@ -82,28 +94,7 @@
{
"channels":
{
"title": "Channels",
"type": "object",
"properties":
{
"sessions":
{
"title": "Kafka Sessions Channel",
"type": "string"
},
"messages":
{
"title": "Kafka Messages Channel",
"type": "string"
},
"retained":
{
"title": "Kafka Retained Channel",
"type": "string"
},
"additionalProperties": false
},
"additionalProperties": false
"$ref": "#/$defs/options/binding/mqtt-kafka/topics"
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm not sure about this one, since we're deliberately using channels asyncapi concept here, whereas mqtt-kafka deals with kafka-specific topics instead, so unifying this config causes asyncapi to switch to topics instead.

}
},
"additionalProperties": false
Expand Down
5 changes: 5 additions & 0 deletions ...yncapi.spec/src/test/java/io/aklivity/zilla/specs/binding/asyncapi/config/SchemaTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,11 @@ public class SchemaTest
@Rule
public final ConfigSchemaRule schema = new ConfigSchemaRule()
.schemaPatch("io/aklivity/zilla/specs/binding/asyncapi/schema/asyncapi.schema.patch.json")
.schemaPatch("io/aklivity/zilla/specs/binding/tls/schema/tls.schema.patch.json")
.schemaPatch("io/aklivity/zilla/specs/binding/tcp/schema/tcp.schema.patch.json")
.schemaPatch("io/aklivity/zilla/specs/binding/kafka/schema/kafka.schema.patch.json")
.schemaPatch("io/aklivity/zilla/specs/binding/http/schema/http.schema.patch.json")
.schemaPatch("io/aklivity/zilla/specs/binding/mqtt/kafka/schema/mqtt.kafka.schema.patch.json")
.schemaPatch("io/aklivity/zilla/specs/engine/schema/vault/test.schema.patch.json")
.configurationRoot("io/aklivity/zilla/specs/binding/asyncapi/config");

Expand Down
5 changes: 5 additions & 0 deletions incubator/binding-asyncapi/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -209,6 +209,11 @@
</artifactItem>
</artifactItems>
<includes>io/aklivity/zilla/specs/binding/asyncapi/schema/asyncapi.schema.patch.json,
io/aklivity/zilla/specs/binding/mqtt/kafka/schema/mqtt.kafka.schema.patch.json,
io/aklivity/zilla/specs/binding/http/schema/http.schema.patch.json,
io/aklivity/zilla/specs/binding/kafka/schema/kafka.schema.patch.json,
io/aklivity/zilla/specs/binding/tcp/schema/tcp.schema.patch.json,
io/aklivity/zilla/specs/binding/tls/schema/tls.schema.patch.json,
io/aklivity/zilla/specs/binding/asyncapi/schema/asyncapi.2.6.0.schema.json,
io/aklivity/zilla/specs/binding/asyncapi/schema/asyncapi.3.0.1.schema.json
</includes>
Expand Down
3 changes: 3 additions & 0 deletions incubator/binding-openapi-asyncapi.spec/NOTICE
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,10 @@ This project includes:
zilla::specs::binding-mqtt-kafka.spec under Aklivity Community License Agreement
zilla::specs::binding-mqtt.spec under The Apache Software License, Version 2.0
zilla::specs::binding-proxy.spec under The Apache Software License, Version 2.0
zilla::specs::binding-tcp.spec under The Apache Software License, Version 2.0
zilla::specs::binding-tls.spec under The Apache Software License, Version 2.0
zilla::specs::engine.spec under The Apache Software License, Version 2.0
zilla::specs::vault-filesystem.spec under The Apache Software License, Version 2.0


This project also includes code under copyright of the following entities:
Expand Down
5 changes: 5 additions & 0 deletions incubator/binding-openapi-asyncapi/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,11 @@
<artifactId>binding-http-kafka</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>io.aklivity.zilla</groupId>
<artifactId>binding-mqtt-kafka</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>io.aklivity.zilla</groupId>
<artifactId>binding-http</artifactId>
Expand Down
3 changes: 3 additions & 0 deletions incubator/binding-openapi.spec/NOTICE
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,10 @@ This project includes:
org.leadpony.justify under The Apache Software License, Version 2.0
zilla::specs::binding-http.spec under The Apache Software License, Version 2.0
zilla::specs::binding-proxy.spec under The Apache Software License, Version 2.0
zilla::specs::binding-tcp.spec under The Apache Software License, Version 2.0
zilla::specs::binding-tls.spec under The Apache Software License, Version 2.0
zilla::specs::engine.spec under The Apache Software License, Version 2.0
zilla::specs::vault-filesystem.spec under The Apache Software License, Version 2.0


This project also includes code under copyright of the following entities:
Expand Down
10 changes: 10 additions & 0 deletions incubator/binding-openapi.spec/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,16 @@
<artifactId>binding-http.spec</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>binding-tls.spec</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>binding-tcp.spec</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
Expand Down
15 changes: 12 additions & 3 deletions ...src/main/scripts/io/aklivity/zilla/specs/binding/openapi/schema/openapi.schema.patch.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,15 +35,24 @@
{
"properties":
{
"tcp": "#/$defs/binding/tcp/options",
"tls": "#/$defs/binding/tls/options",
"tcp":
{
"$ref": "#/$defs/options/binding/tcp"
},
"tls":
{
"$ref": "#/$defs/options/binding/tls"
},
"http":
{
"title": "Http",
"type": "object",
"properties":
{
"authorization": "$defs/options/binding/http/authorization"
"authorization":
{
"$ref": "#/$defs/options/binding/http/authorization"
}
},
"additionalProperties": false
},
Expand Down
11 changes: 11 additions & 0 deletions ...openapi.spec/src/test/java/io/aklivity/zilla/specs/binding/openapi/config/SchemaTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,9 @@ public class SchemaTest
@Rule
public final ConfigSchemaRule schema = new ConfigSchemaRule()
.schemaPatch("io/aklivity/zilla/specs/binding/openapi/schema/openapi.schema.patch.json")
.schemaPatch("io/aklivity/zilla/specs/binding/tls/schema/tls.schema.patch.json")
.schemaPatch("io/aklivity/zilla/specs/binding/tcp/schema/tcp.schema.patch.json")
.schemaPatch("io/aklivity/zilla/specs/binding/http/schema/http.schema.patch.json")
.schemaPatch("io/aklivity/zilla/specs/engine/schema/vault/test.schema.patch.json")
.configurationRoot("io/aklivity/zilla/specs/binding/openapi/config");

Expand All @@ -42,6 +45,14 @@ public void shouldValidateServer()
assertThat(config, not(nullValue()));
}

@Test
public void shouldValidateServerSecure()
{
JsonObject config = schema.validate("server-secure.yaml");

assertThat(config, not(nullValue()));
}

@Test
public void shouldValidateClient()
{
Expand Down
3 changes: 3 additions & 0 deletions incubator/binding-openapi/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -174,6 +174,9 @@
</artifactItem>
</artifactItems>
<includes>io/aklivity/zilla/specs/binding/openapi/schema/openapi.schema.patch.json,
io/aklivity/zilla/specs/binding/http/schema/http.schema.patch.json,
io/aklivity/zilla/specs/binding/tcp/schema/tcp.schema.patch.json,
io/aklivity/zilla/specs/binding/tls/schema/tls.schema.patch.json,
io/aklivity/zilla/specs/binding/openapi/schema/openapi.3.0.3.schema.json,
io/aklivity/zilla/specs/binding/openapi/schema/openapi.3.1.0.schema.json
</includes>
Expand Down
9 changes: 6 additions & 3 deletions .../aklivity/zilla/runtime/binding/mqtt/kafka/internal/stream/MqttKafkaSubscribeFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -692,10 +692,13 @@ private void doMqttBegin(
long authorization,
long affinity)
{
state = MqttKafkaState.openingReply(state);
if (!MqttKafkaState.replyOpening(state))
{
state = MqttKafkaState.openingReply(state);

doBegin(mqtt, originId, routedId, replyId, replySeq, replyAck, replyMax,
traceId, authorization, affinity);
doBegin(mqtt, originId, routedId, replyId, replySeq, replyAck, replyMax,
traceId, authorization, affinity);
}
}

private void doMqttData(
Expand Down
5 changes: 4 additions & 1 deletion ....spec/src/main/scripts/io/aklivity/zilla/specs/binding/http/schema/http.schema.patch.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -147,7 +147,10 @@
}
]
},
"authorization": "$defs/options/binding/http/authorization",
"authorization":
{
"$ref": "#/$defs/options/binding/http/authorization"
},
"overrides":
{
"title": "Overrides",
Expand Down
5 changes: 4 additions & 1 deletion ...pec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/schema/kafka.schema.patch.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -101,7 +101,10 @@
"pattern": "([^\\:]+):(\\d+)"
}
},
"sasl": "$defs/options/binding/kafka/sasl"
"sasl":
{
"$ref": "#/$defs/options/binding/kafka/sasl"
}
},
"additionalProperties": false
},
Expand Down
67 changes: 39 additions & 28 deletions ...in/scripts/io/aklivity/zilla/specs/binding/mqtt/kafka/schema/mqtt.kafka.schema.patch.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,34 +43,7 @@
},
"topics":
{
"title": "Topics",
"type": "object",
"properties":
{
"sessions":
{
"title": "Kafka Sessions Topic",
"type": "string"
},
"messages":
{
"title": "Kafka Messages Topic",
"type": "string"
},
"retained":
{
"title": "Kafka Retained Topic",
"type": "string"
},
"additionalProperties": false
},
"required":
[
"sessions",
"retained",
"messages"
],
"additionalProperties": false
"$ref": "#/$defs/options/binding/mqtt-kafka/topics"
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

See feedback on asyncapi options for mqtt-kafka.

},
"clients":
{
Expand Down Expand Up @@ -188,5 +161,43 @@
]
}
}
},
{
"op": "add",
"path": "/$defs/options/binding/mqtt-kafka",
"value":
{
"topics":
{
"title": "Topics",
"type": "object",
"properties":
{
"sessions":
{
"title": "Kafka Sessions Topic",
"type": "string"
},
"messages":
{
"title": "Kafka Messages Topic",
"type": "string"
},
"retained":
{
"title": "Kafka Retained Topic",
"type": "string"
},
"additionalProperties": false
},
"required":
[
"sessions",
"retained",
"messages"
],
"additionalProperties": false
}
}
}
]
Loading
Loading