Skip to content

Commit 159a0e9

Browse files
authored
Add Pipeline controller and Kafka JDBC driver (#73)
* Add Pipeline controller and Kafka JDBC driver * PR review fixes h/t @jogrogan
1 parent e12144b commit 159a0e9

File tree

36 files changed

+893
-101
lines changed

36 files changed

+893
-101
lines changed

.github/workflows/integration-tests.yml

+2
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,8 @@ jobs:
3838
run: make deploy
3939
- name: Deploy Samples
4040
run: make deploy-samples
41+
- name: Wait for Readiness
42+
run: kubectl wait kafka/one --for=condition=Ready --timeout=10m -n kafka
4143
- name: Run Integration Tests
4244
run: make integration-tests
4345
- name: Capture Cluster State

Makefile

+7-4
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11

22
install:
3-
./gradlew installDist
3+
./gradlew compileJava installDist
44

55
build:
66
./gradlew build
@@ -9,8 +9,11 @@ build:
99

1010
bounce: build undeploy deploy deploy-samples deploy-config deploy-demo
1111

12+
# Integration tests expect K8s and Kafka to be running
1213
integration-tests:
13-
echo "\nNOTHING TO DO FOR NOW"
14+
kubectl port-forward -n kafka svc/one-kafka-external-0 9092 & echo $$! > port-forward.pid
15+
./gradlew intTest || kill `cat port-forward.pid`
16+
kill `cat port-forward.pid`
1417

1518
clean:
1619
./gradlew clean
@@ -26,7 +29,7 @@ undeploy:
2629
kubectl delete -f ./deploy || echo "skipping"
2730
kubectl delete configmap hoptimator-configmap || echo "skipping"
2831

29-
quickstart: build deploy-dev-environment deploy
32+
quickstart: build deploy
3033

3134
deploy-dev-environment:
3235
kubectl create -f https://github.com/jetstack/cert-manager/releases/download/v1.8.2/cert-manager.yaml || echo "skipping"
@@ -54,4 +57,4 @@ release:
5457
test -n "$(VERSION)" # MISSING ARG: $$VERSION
5558
./gradlew publish
5659

57-
.PHONY: build clean quickstart deploy-dev-environment deploy deploy-samples deploy-config integration-tests bounce generate-models release
60+
.PHONY: build clean quickstart deploy-dev-environment deploy deploy-samples deploy-demo deploy-config integration-tests bounce generate-models release

README.md

+1
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,7 @@ Hoptimator requires a Kubernetes cluster. To connect from outside a Kubernetes c
4242
```
4343
$ make install # build and install SQL CLI
4444
$ make deploy deploy-demo # install CRDs and K8s objects
45+
$ kubectl port-forward -n kafka svc/one-kafka-external-0 9092 &
4546
$ ./hoptimator
4647
> !intro
4748
```

deploy/dev/kafka.yaml

+11-4
Original file line numberDiff line numberDiff line change
@@ -26,14 +26,18 @@ spec:
2626
version: 3.8.0
2727
replicas: 1
2828
listeners:
29-
- name: plain
30-
port: 9092
31-
type: internal
32-
tls: false
3329
- name: tls
3430
port: 9093
3531
type: internal
3632
tls: true
33+
- name: external
34+
port: 9092
35+
type: nodeport
36+
tls: false
37+
configuration:
38+
brokers:
39+
- broker: 0
40+
advertisedHost: localhost
3741
config:
3842
offsets.topic.replication.factor: 1
3943
transaction.state.log.replication.factor: 1
@@ -50,3 +54,6 @@ spec:
5054
replicas: 3
5155
storage:
5256
type: ephemeral
57+
entityOperator:
58+
topicOperator: {}
59+

deploy/samples/kafkadb.yaml

+71
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,71 @@
1+
apiVersion: hoptimator.linkedin.com/v1alpha1
2+
kind: Database
3+
metadata:
4+
name: kafka-database
5+
spec:
6+
schema: KAFKA
7+
url: jdbc:kafka://bootstrap.servers=localhost:9092
8+
dialect: Calcite
9+
10+
---
11+
12+
apiVersion: hoptimator.linkedin.com/v1alpha1
13+
kind: TableTemplate
14+
metadata:
15+
name: kafka-template
16+
spec:
17+
databases:
18+
- kafka-database
19+
yaml: |
20+
apiVersion: kafka.strimzi.io/v1beta2
21+
kind: KafkaTopic
22+
metadata:
23+
name: {{name}}
24+
namespace: kafka
25+
labels:
26+
strimzi.io/cluster: one
27+
spec:
28+
topicName: {{table}}
29+
partitions: 1
30+
replicas: 1
31+
config:
32+
retention.ms: 7200000
33+
segment.bytes: 1073741824
34+
connector: |
35+
connector = kafka
36+
topic = {{table}}
37+
properties.bootstrap.servers = localhost:9092
38+
39+
---
40+
41+
apiVersion: kafka.strimzi.io/v1beta2
42+
kind: KafkaTopic
43+
metadata:
44+
name: existing-topic-1
45+
namespace: kafka
46+
labels:
47+
strimzi.io/cluster: one
48+
spec:
49+
partitions: 1
50+
replicas: 1
51+
config:
52+
retention.ms: 7200000
53+
segment.bytes: 1073741824
54+
55+
---
56+
57+
apiVersion: kafka.strimzi.io/v1beta2
58+
kind: KafkaTopic
59+
metadata:
60+
name: existing-topic-2
61+
namespace: kafka
62+
labels:
63+
strimzi.io/cluster: one
64+
spec:
65+
partitions: 1
66+
replicas: 1
67+
config:
68+
retention.ms: 7200000
69+
segment.bytes: 1073741824
70+
71+

hoptimator-cli/build.gradle

+1
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ dependencies {
1111
implementation project(':hoptimator-avro')
1212
implementation project(':hoptimator-demodb')
1313
implementation project(':hoptimator-jdbc')
14+
implementation project(':hoptimator-kafka')
1415
implementation project(':hoptimator-k8s')
1516
implementation project(':hoptimator-util')
1617
implementation libs.calcite.core

hoptimator-cli/src/main/java/sqlline/HoptimatorAppConfig.java

+6-6
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,7 @@ public Collection<CommandHandler> getCommandHandlers(SqlLine sqlline) {
3636
list.addAll(super.getCommandHandlers(sqlline));
3737
list.add(new IntroCommandHandler(sqlline));
3838
list.add(new PipelineCommandHandler(sqlline));
39-
list.add(new YamlCommandHandler(sqlline));
39+
list.add(new SpecifyCommandHandler(sqlline));
4040
return list;
4141
}
4242

@@ -112,17 +112,17 @@ public boolean echoToFile() {
112112
}
113113

114114

115-
private static final class YamlCommandHandler implements CommandHandler {
115+
private static final class SpecifyCommandHandler implements CommandHandler {
116116

117117
private final SqlLine sqlline;
118118

119-
private YamlCommandHandler(SqlLine sqlline) {
119+
private SpecifyCommandHandler(SqlLine sqlline) {
120120
this.sqlline = sqlline;
121121
}
122122

123123
@Override
124124
public String getName() {
125-
return "yaml";
125+
return "specify";
126126
}
127127

128128
@Override
@@ -137,7 +137,7 @@ public String getHelpText() {
137137

138138
@Override
139139
public String matches(String line) {
140-
if (startsWith(line, "!yaml") || startsWith(line, "yaml")) {
140+
if (startsWith(line, "!spec") || startsWith(line, "spec")) {
141141
return line;
142142
} else {
143143
return null;
@@ -147,7 +147,7 @@ public String matches(String line) {
147147
@Override
148148
public void execute(String line, DispatchCallback dispatchCallback) {
149149
if (!(sqlline.getConnection() instanceof CalciteConnection)) {
150-
sqlline.error("This connection doesn't support `!yaml`.");
150+
sqlline.error("This connection doesn't support `!specify`.");
151151
dispatchCallback.setToFailure();
152152
return;
153153
}

hoptimator-jdbc/build.gradle

+8-1
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
plugins {
22
id 'java'
3+
id 'java-test-fixtures'
34
}
45

56
dependencies {
@@ -8,11 +9,17 @@ dependencies {
89
implementation libs.calcite.core
910
implementation libs.calcite.server
1011
implementation libs.slf4j.api
11-
testImplementation libs.quidem
12+
13+
testFixturesImplementation libs.quidem
14+
testFixturesImplementation libs.calcite.core
15+
testFixturesImplementation project(':hoptimator-api')
16+
testFixturesImplementation project(':hoptimator-util')
1217

1318
testRuntimeOnly project(':hoptimator-demodb')
19+
testFixturesImplementation(platform('org.junit:junit-bom:5.11.3'))
1420
testImplementation(platform('org.junit:junit-bom:5.11.3'))
1521
testImplementation 'org.junit.jupiter:junit-jupiter'
22+
testFixturesImplementation 'org.junit.jupiter:junit-jupiter'
1623
testRuntimeOnly 'org.junit.platform:junit-platform-launcher'
1724
testRuntimeOnly 'org.junit.jupiter:junit-jupiter-engine'
1825
}

hoptimator-jdbc/src/test/java/com/linkedin/hoptimator/jdbc/QuidemTestBase.java

-51
This file was deleted.
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,99 @@
1+
package com.linkedin.hoptimator.jdbc;
2+
3+
import com.linkedin.hoptimator.jdbc.HoptimatorDriver;
4+
import com.linkedin.hoptimator.util.ConnectionService;
5+
import com.linkedin.hoptimator.util.DeploymentService;
6+
import com.linkedin.hoptimator.util.Sink;
7+
import com.linkedin.hoptimator.util.planner.PipelineRel;
8+
9+
import net.hydromatic.quidem.AbstractCommand;
10+
import net.hydromatic.quidem.Command;
11+
import net.hydromatic.quidem.CommandHandler;
12+
import net.hydromatic.quidem.Quidem;
13+
14+
import org.apache.calcite.rel.RelNode;
15+
import org.apache.calcite.jdbc.CalciteConnection;
16+
17+
import org.junit.jupiter.api.AfterAll;
18+
import org.junit.jupiter.api.Assertions;
19+
import org.junit.jupiter.api.BeforeAll;
20+
21+
import java.io.File;
22+
import java.io.FileReader;
23+
import java.io.IOException;
24+
import java.io.PrintWriter;
25+
import java.io.Reader;
26+
import java.io.Writer;
27+
import java.net.URI;
28+
import java.net.URISyntaxException;
29+
import java.nio.file.Files;
30+
import java.nio.charset.StandardCharsets;
31+
import java.util.ArrayList;
32+
import java.util.Arrays;
33+
import java.util.Collections;
34+
import java.util.List;
35+
import java.util.Map;
36+
import java.util.stream.Collectors;
37+
import java.sql.Connection;
38+
import java.sql.DriverManager;
39+
40+
public abstract class QuidemTestBase {
41+
42+
protected void run(String resourceName) throws IOException, URISyntaxException {
43+
run(Thread.currentThread().getContextClassLoader().getResource(resourceName).toURI());
44+
}
45+
46+
protected void run(URI resource) throws IOException {
47+
File in = new File(resource);
48+
File out = File.createTempFile(in.getName(), ".out");
49+
try (Reader r = new FileReader(in);
50+
Writer w = new PrintWriter(out)) {
51+
Quidem.Config config = Quidem.configBuilder()
52+
.withReader(r)
53+
.withWriter(w)
54+
.withConnectionFactory((x, y) -> DriverManager.getConnection("jdbc:hoptimator://" + x))
55+
.withCommandHandler(new CustomCommandHandler())
56+
.build();
57+
new Quidem(config).execute();
58+
}
59+
List<String> input = Files.readAllLines(in.toPath(), StandardCharsets.UTF_8);
60+
List<String> output = Files.readAllLines(out.toPath(), StandardCharsets.UTF_8);
61+
Assertions.assertTrue(!input.isEmpty(), "input script is empty");
62+
Assertions.assertTrue(!output.isEmpty(), "script output is empty");
63+
for (String line : output) {
64+
System.out.println(line);
65+
}
66+
Assertions.assertIterableEquals(input, output);
67+
}
68+
69+
private static class CustomCommandHandler implements CommandHandler {
70+
@Override
71+
public Command parseCommand(List<String> lines, List<String> content, final String line) {
72+
List<String> copy = new ArrayList<>();
73+
copy.addAll(lines);
74+
if (line.startsWith("spec")) {
75+
return new AbstractCommand() {
76+
@Override
77+
public void execute(Context context, boolean execute) throws Exception {
78+
if (execute) {
79+
if (!(context.connection() instanceof CalciteConnection)) {
80+
throw new IllegalArgumentException("This connection doesn't support `!specify`.");
81+
}
82+
String sql = context.previousSqlCommand().sql;
83+
CalciteConnection conn = (CalciteConnection) context.connection();
84+
RelNode rel = HoptimatorDriver.convert(conn.createPrepareContext(), sql).root.rel;
85+
String specs = DeploymentService.plan(rel).pipeline().specify().stream()
86+
.collect(Collectors.joining("\n---\n"));
87+
context.echo(Collections.singletonList(specs));
88+
} else {
89+
context.echo(content);
90+
}
91+
context.echo(copy);
92+
}
93+
};
94+
}
95+
96+
return null;
97+
}
98+
}
99+
}

0 commit comments

Comments
 (0)