Skip to content

Commit bd58cfb

Browse files
maxenglanderPhanaticdependabot[bot]
authored
2.4.0 planetscale v1 (#1)
The Debezium Connector for PlanetScale is built on top the Debezium Connector for Vitess v2.4.0.Final with: * Basic authentication headers, required by PlanetScale's edge. * TLS/mTLS support. * Emit DDL records for Kafka Connect, use historicized schema support. ...and more. Some credit-giving book-keeping: all of the commits co-authored by Phani Raj are entirely his. I just copied them from our private fork now that we're ready to open source these changes. --------- Signed-off-by: Max Englander <[email protected]> Co-authored-by: Phani Raj <[email protected]> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
1 parent 82b7cb9 commit bd58cfb

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

52 files changed

+2137
-365
lines changed

‎.github/workflows/contributor-check.yml

+2-2
Original file line numberDiff line numberDiff line change
@@ -37,14 +37,14 @@ jobs:
3737
done < AUTHOR_NAME.txt
3838
- name: Create comment
3939
if: ${{ steps.check.outputs.USER_NOT_FOUND == 'true' }}
40-
uses: peter-evans/create-or-update-comment@v3
40+
uses: peter-evans/create-or-update-comment@v4
4141
with:
4242
issue-number: ${{ github.event.pull_request.number }}
4343
body: |
4444
Welcome as a new contributor to Debezium, @${{ github.event.pull_request.user.login }}. Reviewers, please add missing author name(s) and alias name(s) to the [COPYRIGHT.txt](https://github.com/debezium/debezium/blob/main/COPYRIGHT.txt) and [Aliases.txt](https://github.com/debezium/debezium/blob/main/jenkins-jobs/scripts/config/Aliases.txt) respectively.
4545
- name: Check failure
4646
if: ${{ steps.check.outputs.USER_NOT_FOUND == 'true' }}
47-
uses: actions/github-script@v6
47+
uses: actions/github-script@v7
4848
continue-on-error: false
4949
with:
5050
script: |

‎.github/workflows/cross-maven.yml

+6-3
Original file line numberDiff line numberDiff line change
@@ -53,15 +53,18 @@ jobs:
5353
uses: actions/checkout@v4
5454
with:
5555
repository: debezium/debezium
56-
ref: ${{ github.base_ref }}
56+
#ref: ${{ github.base_ref }}
57+
# XXX(maxenglander): all commits in the 2.4.0-planetscale branch
58+
# descend from upstream v2.4.0.Final.
59+
ref: v2.4.0.Final
5760
path: core
5861
- name: Set up JDK
59-
uses: actions/setup-java@v3
62+
uses: actions/setup-java@v4
6063
with:
6164
distribution: 'temurin'
6265
java-version: 17
6366
- name: Cache Maven packages
64-
uses: actions/cache@v3
67+
uses: actions/cache@v4
6568
with:
6669
path: ~/.m2
6770
key: ${{ runner.os }}-m2-${{ hashFiles('**/pom.xml') }}

‎.github/workflows/maven.yml

+6-3
Original file line numberDiff line numberDiff line change
@@ -23,19 +23,22 @@ jobs:
2323
uses: actions/checkout@v4
2424
with:
2525
repository: debezium/debezium
26-
ref: ${{ github.ref }}
26+
#ref: ${{ github.ref }}
27+
# XXX(maxenglander): all the commits in the 2.4.0-planetscale branch
28+
# descend from upstream 2.4.0.Final.
29+
ref: v2.4.0.Final
2730
path: core
2831
- name: Checkout Vitess connector
2932
uses: actions/checkout@v4
3033
with:
3134
path: vitess
3235
- name: Set up JDK
33-
uses: actions/setup-java@v3
36+
uses: actions/setup-java@v4
3437
with:
3538
distribution: 'temurin'
3639
java-version: 17
3740
- name: Cache Maven packages
38-
uses: actions/cache@v3
41+
uses: actions/cache@v4
3942
with:
4043
path: ~/.m2
4144
key: ${{ runner.os }}-m2-${{ hashFiles('**/pom.xml') }}

‎.github/workflows/octocat-commit-check.yml

+2-2
Original file line numberDiff line numberDiff line change
@@ -28,15 +28,15 @@ jobs:
2828
done < SORTED_AUTHORS.txt
2929
- name: Create comment
3030
if: ${{ steps.octocat.outputs.OCTOCAT_COMMIT_FOUND == 'true' }}
31-
uses: peter-evans/create-or-update-comment@v3
31+
uses: peter-evans/create-or-update-comment@v4
3232
with:
3333
issue-number: ${{ github.event.pull_request.number }}
3434
body: |
3535
Hi @${{ github.event.pull_request.user.login }}. Thank you for your valuable contribution.
3636
Please author your commit(s) using an email [linked to your GitHub account](https://docs.github.com/en/account-and-profile/setting-up-and-managing-your-github-profile/managing-contribution-graphs-on-your-profile/why-are-my-contributions-not-showing-up-on-my-profile).
3737
- name: Check failure
3838
if: ${{ steps.check.outputs.OCTOCAT_COMMIT_FOUND == 'true' }}
39-
uses: actions/github-script@v6
39+
uses: actions/github-script@v7
4040
continue-on-error: false
4141
with:
4242
script: |

‎.github/workflows/sanity-check.yml

-53
This file was deleted.

‎.gitignore

+2
Original file line numberDiff line numberDiff line change
@@ -28,3 +28,5 @@ generated-sources/
2828

2929
/state/
3030
bin/
31+
32+
history.dat

‎pom.xml

+85
Original file line numberDiff line numberDiff line change
@@ -136,17 +136,59 @@
136136
<type>test-jar</type>
137137
<scope>test</scope>
138138
</dependency>
139+
140+
<dependency>
141+
<groupId>io.debezium</groupId>
142+
<artifactId>debezium-storage-file</artifactId>
143+
<version>${version.debezium}</version>
144+
<type>test-jar</type>
145+
<scope>test</scope>
146+
</dependency>
147+
139148
<dependency>
140149
<groupId>io.debezium</groupId>
141150
<artifactId>debezium-embedded</artifactId>
142151
<version>${version.debezium}</version>
143152
<type>test-jar</type>
144153
<scope>test</scope>
145154
</dependency>
155+
156+
<dependency>
157+
<groupId>io.debezium</groupId>
158+
<artifactId>debezium-connector-mysql</artifactId>
159+
<version>${version.debezium}</version>
160+
<type>test-jar</type>
161+
<scope>test</scope>
162+
</dependency>
146163
</dependencies>
147164
</dependencyManagement>
148165

149166
<dependencies>
167+
<!-- SQL Parser -->
168+
<dependency>
169+
<groupId>com.github.jsqlparser</groupId>
170+
<artifactId>jsqlparser</artifactId>
171+
<version>4.7</version>
172+
</dependency>
173+
174+
<dependency>
175+
<groupId>io.debezium</groupId>
176+
<artifactId>debezium-storage-file</artifactId>
177+
<version>${version.debezium}</version>
178+
</dependency>
179+
180+
<dependency>
181+
<groupId>io.debezium</groupId>
182+
<artifactId>debezium-storage-kafka</artifactId>
183+
<version>${version.debezium}</version>
184+
</dependency>
185+
186+
<dependency>
187+
<groupId>io.debezium</groupId>
188+
<artifactId>debezium-connector-mysql</artifactId>
189+
<version>${version.debezium}</version>
190+
</dependency>
191+
150192
<!-- Kafka-->
151193
<dependency>
152194
<groupId>org.apache.kafka</groupId>
@@ -395,6 +437,32 @@
395437
<includeTestSourceDirectory>true</includeTestSourceDirectory>
396438
</configuration>
397439
</plugin>
440+
441+
<plugin>
442+
<groupId>io.confluent</groupId>
443+
<version>0.12.0</version>
444+
<artifactId>kafka-connect-maven-plugin</artifactId>
445+
<executions>
446+
<execution>
447+
<goals>
448+
<goal>kafka-connect</goal>
449+
</goals>
450+
<configuration>
451+
<ownerUsername>planetscale</ownerUsername>
452+
<componentTypes>
453+
<componentType>source</componentType>
454+
</componentTypes>
455+
</configuration>
456+
</execution>
457+
</executions>
458+
<dependencies>
459+
<dependency>
460+
<groupId>org.codehaus.plexus</groupId>
461+
<artifactId>plexus-archiver</artifactId>
462+
<version>3.4</version>
463+
</dependency>
464+
</dependencies>
465+
</plugin>
398466
</plugins>
399467

400468
<resources>
@@ -450,13 +518,30 @@
450518
</goals>
451519
<configuration>
452520
<finalName>${project.artifactId}-${project.version}</finalName>
521+
<appendAssemblyId>false</appendAssemblyId>
453522
<attach>true</attach> <!-- we want attach & deploy these to Maven -->
454523
<descriptorRefs>
455524
<descriptorRef>connector-distribution</descriptorRef>
456525
</descriptorRefs>
457526
<tarLongFileMode>posix</tarLongFileMode>
458527
</configuration>
459528
</execution>
529+
<execution>
530+
<id>jar-with-dependencies</id>
531+
<phase>package</phase>
532+
<goals>
533+
<goal>single</goal>
534+
</goals>
535+
<configuration>
536+
<finalName>${project.artifactId}-${project.version}</finalName>
537+
<appendAssemblyId>true</appendAssemblyId>
538+
<attach>true</attach> <!-- we want attach & deploy these to Maven -->
539+
<descriptorRefs>
540+
<descriptorRef>jar-with-dependencies</descriptorRef>
541+
</descriptorRefs>
542+
<tarLongFileMode>posix</tarLongFileMode>
543+
</configuration>
544+
</execution>
460545
</executions>
461546
</plugin>
462547
</plugins>

‎src/main/java/io/debezium/connector/vitess/VStreamCopyCompletedEventException.java

+12
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,12 @@
1+
/*
2+
* Copyright Debezium Authors.
3+
*
4+
* Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
5+
*/
6+
package io.debezium.connector.vitess;
7+
8+
/**
9+
* Used to signal that Debezium should not continue consuming the streaming event source after encountering a COPY_COMPLETED VEvent from the VStream and when snapshot.mode=initial_only.
10+
*/
11+
public class VStreamCopyCompletedEventException extends RuntimeException {
12+
}

‎src/main/java/io/debezium/connector/vitess/VitessChangeRecordEmitter.java

+10
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121
import io.debezium.relational.RelationalChangeRecordEmitter;
2222
import io.debezium.relational.Table;
2323
import io.debezium.relational.TableId;
24+
import io.debezium.relational.TableSchema;
2425
import io.debezium.util.Clock;
2526
import io.debezium.util.Strings;
2627

@@ -62,6 +63,8 @@ public Envelope.Operation getOperation() {
6263
return Envelope.Operation.UPDATE;
6364
case DELETE:
6465
return Envelope.Operation.DELETE;
66+
case TRUNCATE:
67+
return Envelope.Operation.TRUNCATE;
6568
default:
6669
throw new IllegalArgumentException(
6770
"Received event of unexpected command type: " + message.getOperation());
@@ -72,6 +75,7 @@ public Envelope.Operation getOperation() {
7275
protected Object[] getOldColumnValues() {
7376
switch (getOperation()) {
7477
case CREATE:
78+
case TRUNCATE:
7579
return null;
7680
default:
7781
// UPDATE and DELETE have old values
@@ -91,6 +95,12 @@ protected Object[] getNewColumnValues() {
9195
}
9296
}
9397

98+
@Override
99+
protected void emitTruncateRecord(Receiver receiver, TableSchema tableSchema) throws InterruptedException {
100+
Struct envelope = tableSchema.getEnvelopeSchema().truncate(getOffset().getSourceInfo(), getClock().currentTimeAsInstant());
101+
receiver.changeRecord(getPartition(), tableSchema, Envelope.Operation.TRUNCATE, null, envelope, getOffset(), null);
102+
}
103+
94104
private Object[] columnValues(List<ReplicationMessage.Column> columns, TableId tableId) {
95105
if (columns == null || columns.isEmpty()) {
96106
return null;

0 commit comments

Comments
 (0)