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

Support pgsql-kafka binding #1245

Merged
merged 23 commits into from
Sep 18, 2024
Merged

Support pgsql-kafka binding #1245

Show file tree
Hide file tree
Changes from 16 commits
Commits
Show all changes
23 commits
Select commit Hold shift + click to select a range
e9f3cd4
WIP
akrambek Sep 11, 2024
0474b73
WIP
akrambek Sep 11, 2024
1ddbfbe
WIP
akrambek Sep 12, 2024
d9d9dea
Fix typo
akrambek Sep 12, 2024
e20a21c
Merge branch 'develop' into feature/pgsql-kafka
akrambek Sep 17, 2024
df7700c
WIP
akrambek Sep 17, 2024
e1ee9f7
WIP
akrambek Sep 17, 2024
31b61d6
WIP
akrambek Sep 17, 2024
5daffb7
WIP
akrambek Sep 17, 2024
2956124
WIP No error
akrambek Sep 17, 2024
04c6139
WIP
akrambek Sep 17, 2024
ebe8896
WIP
akrambek Sep 17, 2024
5694fa1
Merge branch 'develop' into feature/pgsql-kafka
akrambek Sep 18, 2024
72d4672
WIP
akrambek Sep 18, 2024
45d8455
WIP
akrambek Sep 18, 2024
ff215d2
WIP
akrambek Sep 18, 2024
8c1436f
WIP
akrambek Sep 18, 2024
05b531e
WIP
akrambek Sep 18, 2024
fc44ecf
WIP
akrambek Sep 18, 2024
865ad2c
Remove avro dependency
akrambek Sep 18, 2024
862d291
Applied feedback from PR
akrambek Sep 18, 2024
8ea8b04
Verify schema registry
akrambek Sep 18, 2024
c950974
include database name in the topic name
akrambek Sep 18, 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-pgsql-kafka.spec/NOTICE
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,5 +14,8 @@ This project includes:
ICU4J under Unicode/ICU License
Jakarta JSON Processing API under Eclipse Public License 2.0 or GNU General Public License, version 2 with the GNU Classpath Exception
org.leadpony.justify under The Apache Software License, Version 2.0
zilla::incubator::binding-pgsql.spec under Aklivity Community License Agreement
zilla::specs::binding-kafka.spec under The Apache Software License, Version 2.0
zilla::specs::binding-proxy.spec under The Apache Software License, Version 2.0
zilla::specs::engine.spec under The Apache Software License, Version 2.0

10 changes: 10 additions & 0 deletions incubator/binding-pgsql-kafka.spec/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,16 @@
<artifactId>engine.spec</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>binding-pgsql.spec</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>binding-kafka.spec</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
Expand Down
40 changes: 37 additions & 3 deletions ...scripts/io/aklivity/zilla/specs/binding/pgsql/kafka/streams/kafka/create.topic/client.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,40 @@
# specific language governing permissions and limitations under the License.
#

connect "zilla://streams/app1"
option zilla:window 8192
option zilla:transmission "half-duplex"

write zilla:begin.ext ${kafka:beginEx()
.typeId(zilla:id("kafka"))
.request()
.describeCluster()
.includeAuthorizedOperations("false")
.build()
.build()}

connected

read zilla:begin.ext ${kafka:matchBeginEx()
.typeId(zilla:id("kafka"))
.response()
.describeCluster()
.throttle(0)
.error(0)
.clusterId("cluster-0")
.controllerId(0)
.broker()
.brokerId(1)
.host("broker1.example.com")
.port(9092)
.build()
.authorizedOperations(0)
.build()
.build()}

write close
read closed

connect "zilla://streams/app1"
option zilla:window 8192
option zilla:transmission "half-duplex"
Expand All @@ -25,10 +59,10 @@ write zilla:begin.ext ${kafka:beginEx()
.name("dev.cities")
.partitionCount(1)
.replicas(1)
.assignment(0, 0)
.assignment(0, 1)
.config("cleanup.policy", "compact")
.build()
.timeout(0)
.timeout(30000)
.validateOnly("false")
.build()
.build()}
Expand All @@ -47,5 +81,5 @@ read zilla:begin.ext ${kafka:matchBeginEx()
.build()
.build()}

read closed
write close
read closed
41 changes: 38 additions & 3 deletions ...scripts/io/aklivity/zilla/specs/binding/pgsql/kafka/streams/kafka/create.topic/server.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,39 @@ accept ${serverAddress}

accepted

read zilla:begin.ext ${kafka:matchBeginEx()
.typeId(zilla:id("kafka"))
.request()
.describeCluster()
.includeAuthorizedOperations("false")
.build()
.build()}

connected

write zilla:begin.ext ${kafka:beginEx()
.typeId(zilla:id("kafka"))
.response()
.describeCluster()
.throttle(0)
.error(0)
.clusterId("cluster-0")
.controllerId(0)
.broker()
.brokerId(1)
.host("broker1.example.com")
.port(9092)
.build()
.authorizedOperations(0)
.build()
.build()}
write flush

read closed
write close

accepted

read zilla:begin.ext ${kafka:matchBeginEx()
.typeId(zilla:id("kafka"))
.request()
Expand All @@ -29,10 +62,10 @@ read zilla:begin.ext ${kafka:matchBeginEx()
.name("dev.cities")
.partitionCount(1)
.replicas(1)
.assignment(0, 0)
.assignment(0, 1)
.config("cleanup.policy", "compact")
.build()
.timeout(0)
.timeout(30000)
.validateOnly("false")
.build()
.build()}
Expand All @@ -51,5 +84,7 @@ write zilla:begin.ext ${kafka:beginEx()
.build()
.build()}

write close
write flush

read closed
write close
4 changes: 2 additions & 2 deletions ...n/scripts/io/aklivity/zilla/specs/binding/pgsql/kafka/streams/kafka/drop.topic/client.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@ write zilla:begin.ext ${kafka:beginEx()
.request()
.deleteTopics()
.topic("dev.cities")
.timeout(0)
.timeout(30000)
.build()
.build()}

Expand All @@ -40,5 +40,5 @@ read zilla:begin.ext ${kafka:matchBeginEx()
.build()
.build()}

read closed
write close
read closed
5 changes: 3 additions & 2 deletions ...n/scripts/io/aklivity/zilla/specs/binding/pgsql/kafka/streams/kafka/drop.topic/server.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ read zilla:begin.ext ${kafka:matchBeginEx()
.request()
.deleteTopics()
.topic("dev.cities")
.timeout(0)
.timeout(30000)
.build()
.build()}

Expand All @@ -43,6 +43,7 @@ write zilla:begin.ext ${kafka:beginEx()
.build()
.build()
.build()}
write flush

write close
read closed
write close
2 changes: 1 addition & 1 deletion ...scripts/io/aklivity/zilla/specs/binding/pgsql/kafka/streams/pgsql/create.topic/client.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,5 +52,5 @@ read advised zilla:flush ${pgsql:flushEx()
.build()
.build()}

read closed
write close
read closed
2 changes: 2 additions & 0 deletions ...scripts/io/aklivity/zilla/specs/binding/pgsql/kafka/streams/pgsql/create.topic/server.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,3 +53,5 @@ write advise zilla:flush ${pgsql:flushEx()
.status("IDLE")
.build()
.build()}
read closed
write close
6 changes: 4 additions & 2 deletions ...n/scripts/io/aklivity/zilla/specs/binding/pgsql/kafka/streams/pgsql/drop.topic/server.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@
# specific language governing permissions and limitations under the License.
#

accept "zilla://streams/app1"
accept "zilla://streams/app0"
option zilla:window 8192
option zilla:transmission "duplex"

Expand Down Expand Up @@ -44,11 +44,13 @@ write advise zilla:flush ${pgsql:flushEx()
.build()
.build()}

write notify CREATE_TOPIC_COMPLETED

write advise zilla:flush ${pgsql:flushEx()
.typeId(zilla:id("pgsql"))
.ready()
.status("IDLE")
.build()
.build()}

write close
read closed
59 changes: 59 additions & 0 deletions ...kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/pgsql/kafka/streams/KafkaIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,59 @@
/*
* Copyright 2021-2023 Aklivity Inc
*
* Licensed under the Aklivity Community License (the "License"); you may not use
* this file except in compliance with the License. You may obtain a copy of the
* License at
*
* https://www.aklivity.io/aklivity-community-license/
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OF ANY KIND, either express or implied. See the License for the
* specific language governing permissions and limitations under the License.
*/
package io.aklivity.zilla.specs.binding.pgsql.kafka.streams;

import static java.util.concurrent.TimeUnit.SECONDS;
import static org.junit.rules.RuleChain.outerRule;

import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.DisableOnDebug;
import org.junit.rules.TestRule;
import org.junit.rules.Timeout;

import io.aklivity.k3po.runtime.junit.annotation.Specification;
import io.aklivity.k3po.runtime.junit.rules.K3poRule;

public class KafkaIT
{
private final K3poRule k3po = new K3poRule()
.addScriptRoot("app", "io/aklivity/zilla/specs/binding/pgsql/kafka/streams/kafka");

private final TestRule timeout = new DisableOnDebug(new Timeout(10, SECONDS));

@Rule
public final TestRule chain = outerRule(k3po).around(timeout);

@Test
@Specification({
"${app}/create.topic/client",
"${app}/create.topic/server"
})
public void shouldCreateTopic() throws Exception
{
k3po.finish();
}

@Test
@Specification({
"${app}/drop.topic/client",
"${app}/drop.topic/server"
})
public void shouldDropTopic() throws Exception
{
k3po.finish();
}

}
58 changes: 58 additions & 0 deletions ...kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/pgsql/kafka/streams/PgsqlIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,58 @@
/*
* Copyright 2021-2023 Aklivity Inc
*
* Licensed under the Aklivity Community License (the "License"); you may not use
* this file except in compliance with the License. You may obtain a copy of the
* License at
*
* https://www.aklivity.io/aklivity-community-license/
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OF ANY KIND, either express or implied. See the License for the
* specific language governing permissions and limitations under the License.
*/
package io.aklivity.zilla.specs.binding.pgsql.kafka.streams;

import static java.util.concurrent.TimeUnit.SECONDS;
import static org.junit.rules.RuleChain.outerRule;

import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.DisableOnDebug;
import org.junit.rules.TestRule;
import org.junit.rules.Timeout;

import io.aklivity.k3po.runtime.junit.annotation.Specification;
import io.aklivity.k3po.runtime.junit.rules.K3poRule;

public class PgsqlIT
{
private final K3poRule k3po = new K3poRule()
.addScriptRoot("app", "io/aklivity/zilla/specs/binding/pgsql/kafka/streams/pgsql");

private final TestRule timeout = new DisableOnDebug(new Timeout(10, SECONDS));

@Rule
public final TestRule chain = outerRule(k3po).around(timeout);

@Test
@Specification({
"${app}/create.topic/client",
"${app}/create.topic/server"
})
public void shouldCreateTopic() throws Exception
{
k3po.finish();
}

@Test
@Specification({
"${app}/drop.topic/client",
"${app}/drop.topic/server"
})
public void shouldDropTopic() throws Exception
{
k3po.finish();
}
}
1 change: 1 addition & 0 deletions incubator/binding-pgsql-kafka/NOTICE
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,4 +10,5 @@ WARRANTIES OF ANY KIND, either express or implied. See the License for the
specific language governing permissions and limitations under the License.

This project includes:
JSQLParser library under GNU Library or Lesser General Public License (LGPL) V2.1 or The Apache Software License, Version 2.0

11 changes: 8 additions & 3 deletions incubator/binding-pgsql-kafka/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@
</licenses>

<properties>
<jacoco.coverage.ratio>0.89</jacoco.coverage.ratio>
<jacoco.coverage.ratio>0.83</jacoco.coverage.ratio>
<jacoco.missed.count>0</jacoco.missed.count>
</properties>

Expand All @@ -41,6 +41,10 @@
<version>${project.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>com.github.jsqlparser</groupId>
<artifactId>jsqlparser</artifactId>
</dependency>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>engine</artifactId>
Expand Down Expand Up @@ -96,7 +100,7 @@
<artifactId>flyweight-maven-plugin</artifactId>
<version>${project.version}</version>
<configuration>
<scopeNames>core pgsql protocol</scopeNames>
<scopeNames>core pgsql kafka protocol</scopeNames>
<packageName>io.aklivity.zilla.runtime.binding.pgsql.kafka.internal.types</packageName>
</configuration>
<executions>
Expand Down Expand Up @@ -135,7 +139,7 @@
</fileMappers>
</artifactItem>
</artifactItems>
<includes>io/aklivity/zilla/specs/binding/pgsql/schema/pgsql.kafka.schema.patch.json</includes>
<includes>io/aklivity/zilla/specs/binding/pgsql/kafka/schema/pgsql.kafka.schema.patch.json</includes>
<outputDirectory>${project.build.directory}/classes</outputDirectory>
</configuration>
</execution>
Expand Down Expand Up @@ -191,6 +195,7 @@
<configuration>
<excludes>
<exclude>io/aklivity/zilla/runtime/binding/pgsql/kafka/internal/types/**/*.class</exclude>
<exclude>net/sf/jsqlparser/parser/*</exclude>
Copy link
Contributor

Choose a reason for hiding this comment

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

Why is this needed? Does jsqlparser generate some code?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

No but it was causing some IlligalClassFormatException in jacoco

java.lang.instrument.IllegalClassFormatException: Error while instrumenting net/sf/jsqlparser/parser/CCJSqlParserTokenManager with JaCoCo 0.8.11.202310140853/f33756c.
	at org.jacoco.agent.rt.internal_4742761.CoverageTransformer.transform(CoverageTransformer.java:94)
	at java.instrument/java.lang.instrument.ClassFileTransformer.transform(ClassFileTransformer.java:244)
	at java.instrument/sun.instrument.TransformerManager.transform(TransformerManager.java:188)
	at java.instrument/sun.instrument.InstrumentationImpl.transform(InstrumentationImpl.java:541)
	at java.base/java.lang.ClassLoader.defineClass1(Native Method)
	at java.base/java.lang.ClassLoader.defineClass(ClassLoader.java:1013)
	at java.base/java.security.SecureClassLoader.defineClass(SecureClassLoader.java:150)
	at java.base/jdk.internal.loader.BuiltinClassLoader.defineClass(BuiltinClassLoader.java:862)
	at java.base/jdk.internal.loader.BuiltinClassLoader.findClassOnClassPathOrNull(BuiltinClassLoader.java:760)
	at java.base/jdk.internal.loader.BuiltinClassLoader.loadClassOrNull(BuiltinClassLoader.java:681)
	at java.base/jdk.internal.loader.BuiltinClassLoader.loadClass(BuiltinClassLoader.java:639)
	at java.base/jdk.internal.loader.ClassLoaders$AppClassLoader.loadClass(ClassLoaders.java:188)
	at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:521)
	at net.sf.jsqlparser.parser.CCJSqlParser.<init>(CCJSqlParser.java:42028)
	at net.sf.jsqlparser.parser.CCJSqlParserManager.parse(CCJSqlParserManager.java:21)

</excludes>
<rules>
<rule>
Expand Down
Loading