DBZ-4393 Create a Debezium Schema Generator for Debezium connectors

* added an API generator for Debezium connectors and static API definitions for connectors in a separate module
* added Maven plug-in
* added GH workflow for debezium-schema-generator

Co-authored-by: rkerner <rkerner.mobil@gmail.com>
Co-authored-by: Anisha Mohanty <anishamohanty23@gmail.com>

.github/workflows/schema-generator-workflow.yml

@@ -0,0 +1,42 @@
+name: Build Schema Generator
+
+on:
+  push:
+    branches:
+      - main
+      - 1.*
+    paths:
+      - 'support/checkstyle/**'
+      - 'debezium-core/**'
+      - 'debezium-schema-generator/**'
+      - 'debezium-parent/pom.xml'
+      - 'debezium-bom/pom.xml'
+      - 'pom.xml'
+      - '.github/workflows/schema-generator-workflow.yml'
+  pull_request:
+    branches:
+      - main
+      - 1.*
+    paths:
+      - 'support/checkstyle/**'
+      - 'debezium-core/**'
+      - 'debezium-schema-generator/**'
+      - 'debezium-parent/pom.xml'
+      - 'debezium-bom/pom.xml'
+      - 'pom.xml'
+      - '.github/workflows/schema-generator-workflow.yml'
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - name: Cache local Maven repository
+        uses: actions/cache@v2
+        with:
+          path: ~/.m2/repository
+          key: ${{ runner.os }}-maven-${{ hashFiles('**/pom.xml') }}
+          restore-keys: |
+            ${{ runner.os }}-maven-
+      - name: Build Schema Generator
+        run: mvn clean install -B -pl debezium-schema-generator -am -Dformat.formatter.goal=validate -Dformat.imports.goal=check -Dorg.slf4j.simpleLogger.log.org.apache.maven.cli.transfer.Slf4jMavenTransferListener=warn -Dmaven.wagon.http.pool=false -Dmaven.wagon.httpconnectionManager.ttlSeconds=120

debezium-bom/pom.xml

@@ -519,6 +519,13 @@
               <version>${version.okhttp}</version>
             </dependency>
 
+            <!-- Code generators -->
+            <dependency>
+              <groupId>io.smallrye</groupId>
+              <artifactId>smallrye-open-api-core</artifactId>
+              <version>2.1.2</version>
+            </dependency>
+
             <!-- Debezium artifacts -->
             <dependency>
                 <groupId>io.debezium</groupId>
@@ -600,6 +607,11 @@
                 <artifactId>debezium-testing-testcontainers</artifactId>
                 <version>${project.version}</version>
             </dependency>
+            <dependency>
+                <groupId>io.debezium</groupId>
+                <artifactId>debezium-schema-generator</artifactId>
+                <version>${project.version}</version>
+            </dependency>
 
             <!-- Debezium test artifacts -->
             <dependency>

debezium-connector-mongodb/src/main/java/io/debezium/connector/mongodb/MongoDbConnectorMetadata.java

@@ -5,37 +5,20 @@
  */
 package io.debezium.connector.mongodb;
 
-import java.util.Arrays;
-import java.util.List;
-
-import org.apache.kafka.connect.source.SourceConnector;
-
 import io.debezium.config.Field;
-import io.debezium.metadata.AbstractConnectorMetadata;
 import io.debezium.metadata.ConnectorDescriptor;
+import io.debezium.metadata.ConnectorMetadata;
 
-public class MongoDbConnectorMetadata extends AbstractConnectorMetadata {
+public class MongoDbConnectorMetadata implements ConnectorMetadata {
 
     @Override
     public ConnectorDescriptor getConnectorDescriptor() {
-        return new ConnectorDescriptor("mongodb", "Debezium MongoDB Connector", getConnector().version());
+        return new ConnectorDescriptor("mongodb", "Debezium MongoDB Connector", MongoDbConnector.class.getName(), Module.version());
     }
 
     @Override
-    public Field.Set getAllConnectorFields() {
+    public Field.Set getConnectorFields() {
-        return MongoDbConnectorConfig.ALL_FIELDS;
+        return MongoDbConnectorConfig.ALL_FIELDS
+                .filtered(f -> f != MongoDbConnectorConfig.POLL_INTERVAL_SEC && f != MongoDbConnectorConfig.MAX_COPY_THREADS);
     }
-
-    @Override
-    public SourceConnector getConnector() {
-        return new MongoDbConnector();
-    }
-
-    @Override
-    public List<String> deprecatedFieldNames() {
-        return Arrays.asList(
-                MongoDbConnectorConfig.POLL_INTERVAL_SEC.name(),
-                MongoDbConnectorConfig.MAX_COPY_THREADS.name());
-    }
-
 }

debezium-connector-mysql/src/main/java/io/debezium/connector/mysql/MySqlConnectorMetadata.java

@@ -5,33 +5,20 @@
  */
 package io.debezium.connector.mysql;
 
-import java.util.Collections;
-import java.util.List;
-
-import org.apache.kafka.connect.source.SourceConnector;
-
 import io.debezium.config.Field;
-import io.debezium.metadata.AbstractConnectorMetadata;
 import io.debezium.metadata.ConnectorDescriptor;
+import io.debezium.metadata.ConnectorMetadata;
 
-public class MySqlConnectorMetadata extends AbstractConnectorMetadata {
+public class MySqlConnectorMetadata implements ConnectorMetadata {
 
     @Override
     public ConnectorDescriptor getConnectorDescriptor() {
-        return new ConnectorDescriptor("mysql", "Debezium MySQL Connector", getConnector().version());
+        return new ConnectorDescriptor("mysql", "Debezium MySQL Connector", MySqlConnector.class.getName(), Module.version());
     }
 
     @Override
-    public SourceConnector getConnector() {
+    public Field.Set getConnectorFields() {
-        return new MySqlConnector();
+        return MySqlConnectorConfig.ALL_FIELDS
-    }
+                .filtered(f -> f != MySqlConnectorConfig.GTID_NEW_CHANNEL_POSITION);
-
-    @Override
-    public Field.Set getAllConnectorFields() {
-        return MySqlConnectorConfig.ALL_FIELDS;
-    }
-
-    public List<String> deprecatedFieldNames() {
-        return Collections.singletonList(MySqlConnectorConfig.GTID_NEW_CHANNEL_POSITION.name());
     }
 }

debezium-connector-postgres/pom.xml

@@ -242,6 +242,24 @@
                     </systemPropertyVariables>
                 </configuration>
             </plugin>
+            <plugin>
+                <groupId>io.debezium</groupId>
+                <artifactId>debezium-schema-generator</artifactId>
+                <version>${project.version}</version>
+                <executions>
+                    <execution>
+                        <id>generate-connector-metadata</id>
+                        <goals>
+                            <goal>generate-openapi-spec</goal>
+                        </goals>
+                        <phase>prepare-package</phase>
+                        <configuration>
+                            <format>openapi</format>
+                            <outputDirectory>${project.build.directory}/generated-sources</outputDirectory>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
         </plugins>
         <resources>
             <!-- Apply the properties set in the POM to the resource files -->

debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/metadata/PostgresConnectorMetadata.java

@@ -3,28 +3,24 @@
  *
  * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
  */
-package io.debezium.connector.postgresql;
+package io.debezium.connector.postgresql.metadata;
-
-import org.apache.kafka.connect.connector.Connector;
 
 import io.debezium.config.Field;
-import io.debezium.metadata.AbstractConnectorMetadata;
+import io.debezium.connector.postgresql.Module;
+import io.debezium.connector.postgresql.PostgresConnector;
+import io.debezium.connector.postgresql.PostgresConnectorConfig;
 import io.debezium.metadata.ConnectorDescriptor;
+import io.debezium.metadata.ConnectorMetadata;
 
-public class PostgresConnectorMetadata extends AbstractConnectorMetadata {
+public class PostgresConnectorMetadata implements ConnectorMetadata {
 
     @Override
     public ConnectorDescriptor getConnectorDescriptor() {
-        return new ConnectorDescriptor("postgres", "Debezium PostgreSQL Connector", getConnector().version());
+        return new ConnectorDescriptor("postgres", "Debezium PostgreSQL Connector", PostgresConnector.class.getName(), Module.version());
     }
 
     @Override
-    public Connector getConnector() {
+    public Field.Set getConnectorFields() {
-        return new PostgresConnector();
-    }
-
-    @Override
-    public Field.Set getAllConnectorFields() {
         return PostgresConnectorConfig.ALL_FIELDS;
     }
 

debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/metadata/PostgresConnectorMetadataProvider.java

@@ -0,0 +1,17 @@
+/*
+ * Copyright Debezium Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.debezium.connector.postgresql.metadata;
+
+import io.debezium.metadata.ConnectorMetadata;
+import io.debezium.metadata.ConnectorMetadataProvider;
+
+public class PostgresConnectorMetadataProvider implements ConnectorMetadataProvider {
+
+    @Override
+    public ConnectorMetadata getConnectorMetadata() {
+        return new PostgresConnectorMetadata();
+    }
+}

debezium-connector-postgres/src/main/resources/META-INF/services/io.debezium.metadata.ConnectorMetadataProvider

@@ -0,0 +1 @@
+io.debezium.connector.postgresql.metadata.PostgresConnectorMetadataProvider

debezium-core/src/main/java/io/debezium/config/Field.java

@@ -17,12 +17,14 @@
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.Objects;
 import java.util.function.BooleanSupplier;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.IntSupplier;
 import java.util.function.LongSupplier;
+import java.util.function.Predicate;
 import java.util.function.Supplier;
 import java.util.regex.Pattern;
 import java.util.regex.PatternSyntaxException;
@@ -172,6 +174,18 @@ public Set with(Iterable<Field> fields) {
         public java.util.Set<String> allFieldNames() {
             return this.fieldsByName.keySet();
         }
+
+        public Set filtered(Predicate<Field> filter) {
+            LinkedHashSet<Field> filtered = new LinkedHashSet<>();
+
+            for (Entry<String, Field> field : fieldsByName.entrySet()) {
+                if (filter.test(field.getValue())) {
+                    filtered.add(field.getValue());
+                }
+            }
+
+            return new Set(filtered);
+        }
     }
 
     /**

debezium-core/src/main/java/io/debezium/metadata/AbstractConnectorMetadata.java

@@ -1,26 +0,0 @@
-/*
- * Copyright Debezium Authors.
- *
- * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
- */
-package io.debezium.metadata;
-
-import java.util.Collections;
-import java.util.List;
-
-import org.apache.kafka.connect.connector.Connector;
-
-import io.debezium.config.Field;
-
-public abstract class AbstractConnectorMetadata {
-
-    public abstract ConnectorDescriptor getConnectorDescriptor();
-
-    public abstract Connector getConnector();
-
-    public abstract Field.Set getAllConnectorFields();
-
-    public List<String> deprecatedFieldNames() {
-        return Collections.emptyList();
-    }
-}

debezium-core/src/main/java/io/debezium/metadata/ConnectorDescriptor.java

@@ -6,14 +6,31 @@
 package io.debezium.metadata;
 
 public class ConnectorDescriptor {
-    public final String id;
+    private final String id;
-    public final String name;
+    private final String name;
-    public final String version;
+    private final String className;
+    private final String version;
 
-    public ConnectorDescriptor(String id, String name, String version) {
+    public ConnectorDescriptor(String id, String name, String className, String version) {
         this.id = id;
         this.name = name;
+        this.className = className;
         this.version = version;
     }
 
+    public String getId() {
+        return id;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public String getClassName() {
+        return className;
+    }
+
+    public String getVersion() {
+        return version;
+    }
 }

debezium-core/src/main/java/io/debezium/metadata/ConnectorMetadata.java

@@ -0,0 +1,15 @@
+/*
+ * Copyright Debezium Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.debezium.metadata;
+
+import io.debezium.config.Field;
+
+public interface ConnectorMetadata {
+
+    ConnectorDescriptor getConnectorDescriptor();
+
+    Field.Set getConnectorFields();
+}

debezium-core/src/main/java/io/debezium/metadata/ConnectorMetadataProvider.java

@@ -0,0 +1,11 @@
+/*
+ * Copyright Debezium Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.debezium.metadata;
+
+public interface ConnectorMetadataProvider {
+
+    ConnectorMetadata getConnectorMetadata();
+}

debezium-parent/pom.xml

@@ -146,6 +146,11 @@
                         <removeUnused>true</removeUnused>
                     </configuration>
                 </plugin>
+                <plugin>
+                    <groupId>org.codehaus.mojo</groupId>
+                    <artifactId>exec-maven-plugin</artifactId>
+                    <version>3.0.0</version>
+                </plugin>
                 <plugin>
                     <groupId>org.jboss.jandex</groupId>
                     <artifactId>jandex-maven-plugin</artifactId>

debezium-schema-generator/pom.xml

@@ -0,0 +1,80 @@
+<?xml version="1.0"?>
+<project xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd" xmlns="http://maven.apache.org/POM/4.0.0"
+    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
+  <modelVersion>4.0.0</modelVersion>
+
+  <parent>
+    <groupId>io.debezium</groupId>
+    <artifactId>debezium-parent</artifactId>
+    <version>1.8.0-SNAPSHOT</version>
+    <relativePath>../debezium-parent/pom.xml</relativePath>
+  </parent>
+
+  <artifactId>debezium-schema-generator</artifactId>
+  <name>Debezium Schema Generator</name>
+  <packaging>maven-plugin</packaging>
+
+  <properties>
+    <maven.compiler.release>11</maven.compiler.release>
+  </properties>
+
+  <dependencies>
+    <dependency>
+      <groupId>io.debezium</groupId>
+      <artifactId>debezium-core</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.kafka</groupId>
+      <artifactId>kafka-clients</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>io.smallrye</groupId>
+      <artifactId>smallrye-open-api-core</artifactId>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.maven</groupId>
+      <artifactId>maven-plugin-api</artifactId>
+      <version>3.6.0</version>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.maven.plugin-tools</groupId>
+      <artifactId>maven-plugin-annotations</artifactId>
+      <version>3.6.0</version>
+      <scope>provided</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.maven</groupId>
+      <artifactId>maven-project</artifactId>
+      <version>2.2.1</version>
+    </dependency>
+
+    <dependency>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
+      <scope>test</scope>
+    </dependency>
+  </dependencies>
+
+  <build>
+    <pluginManagement>
+      <plugins>
+        <plugin>
+          <groupId>org.apache.maven.plugins</groupId>
+          <artifactId>maven-plugin-plugin</artifactId>
+          <version>3.6.0</version>
+        </plugin>
+      </plugins>
+    </pluginManagement>
+    <resources>
+      <resource>
+        <filtering>true</filtering>
+        <directory>src/main/resources</directory>
+        <includes>
+          <include>*</include>
+          <include>**/*</include>
+        </includes>
+      </resource>
+  </resources>
+  </build>
+</project>

debezium-schema-generator/src/main/java/io/debezium/schemagenerator/JsonSchemaCreatorService.java

@@ -0,0 +1,173 @@
+/*
+ * Copyright Debezium Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.debezium.schemagenerator;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.SortedMap;
+import java.util.TreeMap;
+
+import org.apache.kafka.common.config.ConfigDef;
+import org.eclipse.microprofile.openapi.models.media.Schema;
+
+import io.debezium.config.Field;
+import io.debezium.metadata.ConnectorMetadata;
+import io.debezium.schemagenerator.formats.ApiFormat.FieldFilter;
+import io.smallrye.openapi.api.models.media.SchemaImpl;
+
+public class JsonSchemaCreatorService {
+
+    private final String connectorBaseName;
+    private final String connectorName;
+    private final ConnectorMetadata connectorMetadata;
+    private final FieldFilter fieldFilter;
+    private final List<String> errors = new ArrayList<>();
+
+    public JsonSchemaCreatorService(ConnectorMetadata connectorMetadata, FieldFilter fieldFilter) {
+        this.connectorBaseName = connectorMetadata.getConnectorDescriptor().getId();
+        this.connectorName = connectorBaseName + "-" + connectorMetadata.getConnectorDescriptor().getVersion();
+        this.connectorMetadata = connectorMetadata;
+        this.fieldFilter = fieldFilter;
+    }
+
+    public static class JsonSchemaType {
+        public final Schema.SchemaType schemaType;
+        public final String format;
+
+        public JsonSchemaType(Schema.SchemaType schemaType, String format) {
+            this.schemaType = schemaType;
+            this.format = format;
+        }
+
+        public JsonSchemaType(Schema.SchemaType schemaType) {
+            this.schemaType = schemaType;
+            this.format = null;
+        }
+    }
+
+    public List<String> getErrors() {
+        return errors;
+    }
+
+    private Field checkField(Field field) {
+        String propertyName = field.name();
+
+        if (propertyName.contains("whitelist")
+                || propertyName.contains("blacklist")
+                || propertyName.startsWith("internal.")) {
+            // skip legacy and internal properties
+            return null;
+        }
+
+        if (!fieldFilter.include(field)) {
+            // when a property includeList is specified, skip properties not in the list
+            this.errors.add("[INFO] Skipped property \"" + propertyName
+                    + "\" for connector \"" + connectorName + "\" because it was not in the include list file.");
+            return null;
+        }
+
+        if (null == field.group()) {
+            this.errors.add("[WARN] Missing GroupEntry for property \"" + propertyName
+                    + "\" for connector \"" + connectorName + "\".");
+            return field.withGroup(Field.createGroupEntry(Field.Group.ADVANCED));
+        }
+
+        return field;
+    }
+
+    private static JsonSchemaType toJsonSchemaType(ConfigDef.Type type) {
+        switch (type) {
+            case BOOLEAN:
+                return new JsonSchemaType(Schema.SchemaType.BOOLEAN);
+            case CLASS:
+                return new JsonSchemaType(Schema.SchemaType.STRING, "class");
+            case DOUBLE:
+                return new JsonSchemaType(Schema.SchemaType.NUMBER, "double");
+            case INT:
+            case SHORT:
+                return new JsonSchemaType(Schema.SchemaType.INTEGER, "int32");
+            case LIST:
+                return new JsonSchemaType(Schema.SchemaType.STRING, "list,regex");
+            case LONG:
+                return new JsonSchemaType(Schema.SchemaType.INTEGER, "int64");
+            case PASSWORD:
+                return new JsonSchemaType(Schema.SchemaType.STRING, "password");
+            case STRING:
+                return new JsonSchemaType(Schema.SchemaType.STRING);
+            default:
+                throw new IllegalArgumentException("Unsupported property type: " + type);
+        }
+    }
+
+    public Schema buildConnectorSchema() {
+        Schema schema = new SchemaImpl(connectorName);
+        String connectorVersion = connectorMetadata.getConnectorDescriptor().getVersion();
+        schema.setTitle(connectorMetadata.getConnectorDescriptor().getName());
+        schema.setType(Schema.SchemaType.OBJECT);
+        schema.addExtension("connector-id", connectorBaseName);
+        schema.addExtension("version", connectorVersion);
+        schema.addExtension("className", connectorMetadata.getConnectorDescriptor().getClassName());
+
+        Map<Field.Group, SortedMap<Integer, SchemaImpl>> orderedPropertiesByCategory = new HashMap<>();
+
+        Arrays.stream(Field.Group.values()).forEach(category -> {
+            orderedPropertiesByCategory.put(category, new TreeMap<>());
+        });
+
+        connectorMetadata.getConnectorFields().forEach(field -> {
+            String propertyName = field.name();
+            Field checkedField = checkField(field);
+            if (null != checkedField) {
+                SchemaImpl propertySchema = new SchemaImpl(propertyName);
+                Set<?> allowedValues = checkedField.allowedValues();
+                if (null != allowedValues && !allowedValues.isEmpty()) {
+                    propertySchema.enumeration(new ArrayList<>(allowedValues));
+                }
+                if (checkedField.isRequired()) {
+                    propertySchema.nullable(false);
+                    schema.addRequired(propertyName);
+                }
+                propertySchema.description(checkedField.description());
+                propertySchema.defaultValue(checkedField.defaultValue());
+                JsonSchemaType jsonSchemaType = toJsonSchemaType(checkedField.type());
+                propertySchema.type(jsonSchemaType.schemaType);
+                if (null != jsonSchemaType.format) {
+                    propertySchema.format(jsonSchemaType.format);
+                }
+                propertySchema.title(checkedField.displayName());
+                Map<String, Object> extensions = new HashMap<>();
+                extensions.put("name", checkedField.name()); // @TODO remove "x-name" in favor of map key?
+                Field.GroupEntry groupEntry = checkedField.group();
+                extensions.put("category", groupEntry.getGroup().name());
+                propertySchema.extensions(extensions);
+                SortedMap<Integer, SchemaImpl> groupProperties = orderedPropertiesByCategory.get(groupEntry.getGroup());
+                if (groupProperties.containsKey(groupEntry.getPositionInGroup())) {
+                    errors.add("[ERROR] Position in group \"" + groupEntry.getGroup().name() + "\" for property \""
+                            + propertyName + "\" is used more than once for connector \"" + connectorName + "\".");
+                }
+                else {
+                    groupProperties.put(groupEntry.getPositionInGroup(), propertySchema);
+                }
+            }
+        });
+
+        Arrays.stream(Field.Group.values()).forEach(
+                group -> orderedPropertiesByCategory.get(group).forEach((position, propertySchema) -> schema.addProperty(propertySchema.getName(), propertySchema)));
+
+        // Allow additional properties until OAS 3.1 is not avaialble with Swagger/microprofile-openapi
+        // We need JSON Schema `patternProperties`, defined here: https://json-schema.org/understanding-json-schema/reference/object.html#pattern-properties
+        // previously added to OAS 3.1: https://github.com/OAI/OpenAPI-Specification/pull/2489
+        // see https://github.com/eclipse/microprofile-open-api/issues/333
+        // see https://github.com/swagger-api/swagger-core/issues/3913
+        schema.additionalPropertiesBoolean(true);
+
+        return schema;
+    }
+}

debezium-schema-generator/src/main/java/io/debezium/schemagenerator/OpenApiGenerator.java

@@ -0,0 +1,78 @@
+/*
+ * Copyright Debezium Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.debezium.schemagenerator;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Optional;
+import java.util.ServiceLoader;
+import java.util.ServiceLoader.Provider;
+import java.util.stream.Collectors;
+
+import org.eclipse.microprofile.openapi.models.media.Schema;
+
+import com.google.common.base.Charsets;
+import com.google.common.io.Files;
+
+import io.debezium.metadata.ConnectorMetadata;
+import io.debezium.metadata.ConnectorMetadataProvider;
+import io.debezium.schemagenerator.formats.ApiFormat;
+import io.debezium.schemagenerator.formats.ApiFormatName;
+
+public class OpenApiGenerator {
+
+    public static void main(String[] args) {
+        if (args.length != 2) {
+            throw new IllegalArgumentException("Usage: OpenApiGenerator <format-name> <output-directory>");
+        }
+
+        String formatName = args[0].trim();
+        Path outputDirectory = new File(args[1]).toPath();
+
+        new OpenApiGenerator().run(formatName, outputDirectory);
+    }
+
+    private void run(String formatName, Path outputDirectory) {
+        List<ConnectorMetadata> allMetadata = getMetadata();
+
+        ApiFormat format = getApiFormat(formatName);
+
+        for (ConnectorMetadata connectorMetadata : allMetadata) {
+            JsonSchemaCreatorService jsonSchemaCreatorService = new JsonSchemaCreatorService(connectorMetadata, format.getFieldFilter());
+            Schema buildConnectorSchema = jsonSchemaCreatorService.buildConnectorSchema();
+            String spec = format.getSpec(buildConnectorSchema);
+
+            try {
+                Files.write(spec.getBytes(Charsets.UTF_8), outputDirectory.resolve(connectorMetadata.getConnectorDescriptor().getId() + ".json").toFile());
+            }
+            catch (IOException e) {
+                throw new RuntimeException("Couldn't write file", e);
+            }
+        }
+    }
+
+    private List<ConnectorMetadata> getMetadata() {
+        ServiceLoader<ConnectorMetadataProvider> metadataProviders = ServiceLoader.load(ConnectorMetadataProvider.class);
+
+        return metadataProviders.stream()
+                .map(p -> p.get().getConnectorMetadata())
+                .collect(Collectors.toList());
+    }
+
+    /**
+     * Returns the {@link ApiFormat} with the given name, specified via the {@link ApiFormatName} annotation.
+     */
+    private ApiFormat getApiFormat(String formatName) {
+        Optional<Provider<ApiFormat>> format = ServiceLoader.load(ApiFormat.class)
+                .stream()
+                .filter(p -> p.type().getAnnotation(ApiFormatName.class).value().equals(formatName))
+                .findFirst();
+
+        return format.orElseThrow().get();
+    }
+}

debezium-schema-generator/src/main/java/io/debezium/schemagenerator/formats/ApiFormat.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright Debezium Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.debezium.schemagenerator.formats;
+
+import java.util.Map;
+
+import org.eclipse.microprofile.openapi.models.media.Schema;
+
+import io.debezium.config.Field;
+
+public interface ApiFormat {
+
+    ApiFormatDescriptor getDescriptor();
+
+    void configure(Map<String, Object> config);
+
+    String getSpec(Schema connectorSchema);
+
+    /**
+     * Returns a filter to be applied to the fields of the schema. Only matching
+     * fields will be part of the serialized API spec. Defaults to including all the
+     * fields of the schema.
+     */
+    default FieldFilter getFieldFilter() {
+        return f -> true;
+    }
+
+    public interface FieldFilter {
+        boolean include(Field field);
+    }
+
+}

debezium-schema-generator/src/main/java/io/debezium/schemagenerator/formats/ApiFormatDescriptor.java

@@ -0,0 +1,17 @@
+/*
+ * Copyright Debezium Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.debezium.schemagenerator.formats;
+
+public interface ApiFormatDescriptor {
+
+    String getId();
+
+    String getName();
+
+    String getVersion();
+
+    String getDescription();
+}

debezium-schema-generator/src/main/java/io/debezium/schemagenerator/formats/ApiFormatName.java

@@ -0,0 +1,14 @@
+/*
+ * Copyright Debezium Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.debezium.schemagenerator.formats;
+
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+
+@Retention(RetentionPolicy.RUNTIME)
+public @interface ApiFormatName {
+    String value();
+}

debezium-schema-generator/src/main/java/io/debezium/schemagenerator/formats/OpenApiFormat.java

@@ -0,0 +1,97 @@
+/*
+ * Copyright Debezium Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.debezium.schemagenerator.formats;
+
+import java.io.IOException;
+import java.util.Map;
+
+import org.eclipse.microprofile.openapi.models.Components;
+import org.eclipse.microprofile.openapi.models.OpenAPI;
+import org.eclipse.microprofile.openapi.models.media.Schema;
+
+import io.debezium.util.IoUtil;
+import io.smallrye.openapi.api.constants.OpenApiConstants;
+import io.smallrye.openapi.api.models.ComponentsImpl;
+import io.smallrye.openapi.api.models.OpenAPIImpl;
+import io.smallrye.openapi.api.models.info.InfoImpl;
+import io.smallrye.openapi.runtime.io.Format;
+import io.smallrye.openapi.runtime.io.OpenApiSerializer;
+
+@ApiFormatName("openapi")
+public class OpenApiFormat implements ApiFormat {
+
+    private static final ApiFormatDescriptor DESCRIPTOR = new ApiFormatDescriptor() {
+        @Override
+        public String getId() {
+            return "openapi";
+        }
+
+        @Override
+        public String getName() {
+            return "OpenAPI";
+        }
+
+        @Override
+        public String getVersion() {
+            return "3.0.3";
+        }
+
+        @Override
+        public String getDescription() {
+            return "TBD";
+        }
+    };
+
+    private Format format = Format.JSON;
+
+    @Override
+    public ApiFormatDescriptor getDescriptor() {
+        return DESCRIPTOR;
+    }
+
+    @Override
+    public void configure(Map<String, Object> config) {
+        if (null == config || config.isEmpty()) {
+            return;
+        }
+        config.forEach((property, value) -> {
+            switch (property) {
+                case "format":
+                    format = Format.valueOf((String) value);
+                    break;
+                default:
+                    break;
+            }
+        });
+    }
+
+    @Override
+    public String getSpec(Schema connectorSchema) {
+        OpenAPI debeziumAPI = new OpenAPIImpl();
+        debeziumAPI.setOpenapi(OpenApiConstants.OPEN_API_VERSION);
+
+        Components debeziumConnectorTypeComponents = new ComponentsImpl();
+
+        debeziumAPI.setInfo(new InfoImpl());
+        debeziumAPI.getInfo().setTitle("Generated by Debezium OpenAPI Generator");
+
+        debeziumAPI.getInfo().setVersion(
+                IoUtil.loadProperties(OpenApiFormat.class, "io/debezium/schemagenerator/build.properties").getProperty("version"));
+
+        debeziumConnectorTypeComponents.addSchema(
+                "debezium-" + connectorSchema.getExtensions().get("connector-id") + "-" + connectorSchema.getExtensions().get("version"),
+                connectorSchema);
+
+        debeziumAPI.setComponents(debeziumConnectorTypeComponents);
+
+        try {
+            return OpenApiSerializer.serialize(debeziumAPI, format);
+        }
+        catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
+}

debezium-schema-generator/src/main/java/io/debezium/schemagenerator/maven/OpenApiGeneratorMojo.java

@@ -0,0 +1,121 @@
+/*
+ * Copyright Debezium Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.debezium.schemagenerator.maven;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.maven.artifact.Artifact;
+import org.apache.maven.plugin.AbstractMojo;
+import org.apache.maven.plugin.MojoExecutionException;
+import org.apache.maven.plugin.MojoFailureException;
+import org.apache.maven.plugins.annotations.LifecyclePhase;
+import org.apache.maven.plugins.annotations.Mojo;
+import org.apache.maven.plugins.annotations.Parameter;
+import org.apache.maven.project.MavenProject;
+import org.eclipse.microprofile.openapi.models.OpenAPI;
+import org.jboss.jandex.DotName;
+
+import com.fasterxml.jackson.annotation.JsonView;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
+import com.google.common.base.Charsets;
+
+import io.debezium.DebeziumException;
+import io.debezium.schemagenerator.OpenApiGenerator;
+import io.smallrye.openapi.runtime.io.Format;
+
+/**
+ * Generates the OpenAPI spec for the connector(s) in a project.
+ */
+@Mojo(name = "generate-openapi-spec", defaultPhase = LifecyclePhase.PREPARE_PACKAGE)
+public class OpenApiGeneratorMojo extends AbstractMojo {
+
+    @Parameter(property = "openapi.generator.format")
+    private String format;
+
+    @Parameter(defaultValue = "${project.build.outputDirectory}", required = true)
+    private File outputDirectory;
+
+    /**
+     * Gives access to the Maven project information.
+     */
+    @Parameter(defaultValue = "${project}", required = true, readonly = true)
+    private MavenProject project;
+
+    @Override
+    public void execute() throws MojoExecutionException, MojoFailureException {
+        String classPath = getClassPath();
+
+        try {
+            int result = exec(OpenApiGenerator.class.getName(), classPath, Collections.emptyList(), Arrays.<String> asList(format, outputDirectory.getAbsolutePath()));
+
+            if (result != 0) {
+                throw new MojoExecutionException("Couldn't generate OpenAPI spec; please see the logs for more details");
+            }
+            getLog().info("Generated OpenAPI spec at " + outputDirectory.getAbsolutePath());
+        }
+        catch (IOException | InterruptedException e) {
+            throw new MojoExecutionException("Couldn't generate OpenAPI spec", e);
+        }
+    }
+
+    private int exec(String className, String classPath, List<String> jvmArgs, List<String> args) throws IOException, InterruptedException {
+        String javaHome = System.getProperty("java.home");
+        String javaBin = javaHome + File.separator + "bin" + File.separator + "java";
+
+        List<String> command = new ArrayList<>();
+        command.add(javaBin);
+        command.addAll(jvmArgs);
+        command.add("-cp");
+        command.add(classPath);
+        command.add(className);
+        command.addAll(args);
+
+        ProcessBuilder builder = new ProcessBuilder(command);
+        Process process = builder.inheritIO().start();
+        process.waitFor();
+
+        return process.exitValue();
+    }
+
+    private String getClassPath() {
+        Set<Artifact> artifacts = project.getDependencyArtifacts();
+
+        String classPath = artifacts.stream()
+                .filter(a -> a.getScope().equals(Artifact.SCOPE_COMPILE) || a.getScope().equals(Artifact.SCOPE_PROVIDED))
+                .map(a -> a.getFile().getAbsolutePath())
+                .collect(Collectors.joining(File.pathSeparator));
+
+        classPath += classPathEntryFor(getClass());
+        classPath += classPathEntryFor(OpenAPI.class);
+        classPath += classPathEntryFor(ConfigDef.class);
+        classPath += classPathEntryFor(Format.class);
+        classPath += classPathEntryFor(DebeziumException.class);
+        classPath += classPathEntryFor(JsonProcessingException.class);
+        classPath += classPathEntryFor(YAMLFactory.class);
+        classPath += classPathEntryFor(JsonNode.class);
+        classPath += classPathEntryFor(JsonView.class);
+        classPath += classPathEntryFor(DotName.class);
+        classPath += classPathEntryFor(Charsets.class);
+
+        classPath += File.pathSeparator + project.getArtifact().getFile().getAbsolutePath();
+
+        return classPath;
+    }
+
+    private String classPathEntryFor(Class<?> clazz) {
+        return File.pathSeparator + clazz.getProtectionDomain().getCodeSource().getLocation().toString();
+    }
+}

debezium-schema-generator/src/main/resources/META-INF/services/io.debezium.schemagenerator.formats.ApiFormat

@@ -0,0 +1 @@
+io.debezium.schemagenerator.formats.OpenApiFormat

debezium-schema-generator/src/main/resources/io/debezium/schemagenerator/build.properties

@@ -0,0 +1 @@
++version=${project.version}

pom.xml

@@ -153,6 +153,7 @@
         <module>debezium-server</module>
         <module>debezium-testing</module>
         <module>debezium-connect-rest-extension</module>
+        <module>debezium-schema-generator</module>
     </modules>
 
     <distributionManagement>