MINOR: Add git support for schema compatibility checker (#17684)

Add git support for schema compatibility checker. Pulls in valid schema from remote git trunk branch to check with edited schema in local branch. Adds new option for command line verify-evolution-git which takes in a required file name.

Reviewers: Colin P. McCabe <cmccabe@apache.org>

build.gradle

@@ -1814,6 +1814,13 @@ project(':generator') {
     implementation libs.jacksonDatabind
     implementation libs.jacksonJDK8Datatypes
     implementation libs.jacksonJaxrsJsonProvider
+
+    implementation 'org.eclipse.jgit:org.eclipse.jgit:6.4.0.202211300538-r'
+    // SSH support for JGit based on Apache MINA sshd
+    implementation 'org.eclipse.jgit:org.eclipse.jgit.ssh.apache:6.4.0.202211300538-r'
+    // GPG support for JGit based on BouncyCastle (commit signing)
+    implementation 'org.eclipse.jgit:org.eclipse.jgit.gpg.bc:6.4.0.202211300538-r'
+
     testImplementation libs.junitJupiter
 
     testRuntimeOnly runtimeTestLibs

checkstyle/import-control.xml

@@ -380,6 +380,7 @@
     <allow pkg="net.sourceforge.argparse4j" />
     <allow pkg="org.apache.kafka.message" />
     <allow pkg="org.apache.message" />
+    <allow pkg="org.eclipse.jgit" />
   </subpackage>
 
   <subpackage name="streams">

generator/src/main/java/org/apache/kafka/message/checker/CheckerUtils.java

@@ -22,8 +22,21 @@ import org.apache.kafka.message.MessageGenerator;
 import org.apache.kafka.message.MessageSpec;
 import org.apache.kafka.message.Versions;
 
+import org.eclipse.jgit.api.Git;
+import org.eclipse.jgit.lib.ObjectId;
+import org.eclipse.jgit.lib.ObjectLoader;
+import org.eclipse.jgit.lib.Ref;
+import org.eclipse.jgit.lib.Repository;
+import org.eclipse.jgit.revwalk.RevCommit;
+import org.eclipse.jgit.revwalk.RevTree;
+import org.eclipse.jgit.revwalk.RevWalk;
+import org.eclipse.jgit.treewalk.TreeWalk;
+import org.eclipse.jgit.treewalk.filter.PathFilter;
+
 import java.io.File;
+import java.io.IOException;
 import java.nio.file.Files;
+import java.nio.file.Path;
 import java.nio.file.Paths;
 
 /**
@@ -91,4 +104,45 @@ class CheckerUtils {
             throw new RuntimeException("Unable to parse file as MessageSpec: " + schemaPath, e);
         }
     }
+
+    /**
+     * Return a MessageSpec file give file contents.
+     *
+     * @param contents      The path to read the file from.
+     * @return              The MessageSpec.
+     */
+    static MessageSpec readMessageSpecFromString(String contents) {
+        try {
+            return MessageGenerator.JSON_SERDE.readValue(contents, MessageSpec.class);
+        } catch (Exception e) {
+            throw new RuntimeException("Unable to parse string as MessageSpec: " + contents, e);
+        }
+    }
+
+    /**
+     * Read a MessageSpec file from remote git repo.
+     *
+     * @param filePath   The file to read from remote git repo.
+     * @return                     The file contents.
+     */
+    static String GetDataFromGit(String filePath, Path gitPath) throws IOException {
+        Git git = Git.open(new File(gitPath + "/.git"));
+        Repository repository = git.getRepository();
+        Ref head = git.getRepository().getRefDatabase().firstExactRef("refs/heads/trunk");
+        try (RevWalk revWalk = new RevWalk(repository)) {
+            RevCommit commit = revWalk.parseCommit(head.getObjectId());
+            RevTree tree = commit.getTree();
+            try (TreeWalk treeWalk = new TreeWalk(repository)) {
+                treeWalk.addTree(tree);
+                treeWalk.setRecursive(true);
+                treeWalk.setFilter(PathFilter.create(String.valueOf(Paths.get(filePath.substring(1)))));
+                if (!treeWalk.next()) {
+                    throw new IllegalStateException("Did not find expected file " + filePath.substring(1));
+                }
+                ObjectId objectId = treeWalk.getObjectId(0);
+                ObjectLoader loader = repository.open(objectId);
+                return new String(loader.getBytes(), "UTF-8");
+            }
+        }
+    }
 }

generator/src/main/java/org/apache/kafka/message/checker/MetadataSchemaCheckerTool.java

@@ -25,6 +25,10 @@ import net.sourceforge.argparse4j.inf.Subparsers;
 import net.sourceforge.argparse4j.internal.HelpScreenException;
 
 import java.io.PrintStream;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+
+import static org.apache.kafka.message.checker.CheckerUtils.GetDataFromGit;
 
 public class MetadataSchemaCheckerTool {
     public static void main(String[] args) throws Exception {
@@ -56,6 +60,11 @@ public class MetadataSchemaCheckerTool {
         evolutionVerifierParser.addArgument("--path2", "-2").
             required(true).
             help("The final schema JSON path.");
+        Subparser evolutionGitVerifierParser = subparsers.addParser("verify-evolution-git").
+            help(" Verify that an evolution of a JSON file is valid using git.");
+        evolutionGitVerifierParser.addArgument("--file", "-3").
+            required(true).
+            help("The edited JSON file");
         Namespace namespace;
         if (args.length == 0) {
             namespace = argumentParser.parseArgs(new String[] {"--help"});
@@ -81,6 +90,23 @@ public class MetadataSchemaCheckerTool {
                         ", and path2: " + path2);
                 break;
             }
+            case "verify-evolution-git": {
+                String filePath = "/metadata/src/main/resources/common/metadata/" + namespace.getString("file");
+                Path rootKafkaDirectory = Paths.get("").toAbsolutePath();
+                while (!rootKafkaDirectory.endsWith("kafka")) {
+                    rootKafkaDirectory = rootKafkaDirectory.getParent();
+                    if (rootKafkaDirectory == null) {
+                        throw new RuntimeException("Invalid directory, need to be within the kafka directory");
+                    }
+                }
+                String gitContent = GetDataFromGit(filePath, rootKafkaDirectory);
+                EvolutionVerifier verifier = new EvolutionVerifier(
+                        CheckerUtils.readMessageSpecFromFile(rootKafkaDirectory + filePath),
+                        CheckerUtils.readMessageSpecFromString(gitContent));
+                verifier.verify();
+                writer.println("Successfully verified evolution of file: " + namespace.getString("file"));
+                break;
+            }
             default:
                 throw new RuntimeException("Unknown command " + command);
         }

generator/src/test/java/org/apache/kafka/message/checker/MetadataSchemaCheckerToolTest.java

@@ -26,6 +26,15 @@ import static org.apache.kafka.message.checker.CheckerTestUtils.messageSpecStrin
 import static org.junit.jupiter.api.Assertions.assertEquals;
 
 public class MetadataSchemaCheckerToolTest {
+    @Test
+    public void testVerifyEvolutionGit() throws Exception {
+        try (ByteArrayOutputStream stream = new ByteArrayOutputStream()) {
+            MetadataSchemaCheckerTool.run(new String[]{"verify-evolution-git", "--file", "AbortTransactionRecord.json"}, new PrintStream(stream));
+            assertEquals("Successfully verified evolution of file: AbortTransactionRecord.json",
+                stream.toString().trim());
+        }
+    }
+
     @Test
     public void testSuccessfulParse() throws Exception {
         try (ByteArrayOutputStream stream = new ByteArrayOutputStream()) {