HADOOP-12194. Support for incremental generation in the protoc plugin.

hadoop-common-project/hadoop-common/CHANGES.txt

@@ -719,6 +719,9 @@ Release 2.8.0 - UNRELEASED
     HADOOP-12172. FsShell mkdir -p makes an unnecessary check for the existence
     of the parent. (cnauroth)
 
+    HADOOP-12194. Support for incremental generation in the protoc plugin.
+    (wang)
+
   BUG FIXES
 
     HADOOP-11802: DomainSocketWatcher thread terminates sometimes after there

hadoop-maven-plugins/pom.xml

@@ -47,6 +47,14 @@
       <version>${maven.plugin-tools.version}</version>
       <scope>provided</scope>
     </dependency>
+    <dependency>
+      <groupId>org.codehaus.jackson</groupId>
+      <artifactId>jackson-core-asl</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.codehaus.jackson</groupId>
+      <artifactId>jackson-mapper-asl</artifactId>
+    </dependency>
   </dependencies>
   <build>
     <plugins>

hadoop-maven-plugins/src/main/java/org/apache/hadoop/maven/plugin/protoc/ProtocMojo.java

@@ -22,11 +22,21 @@
 import org.apache.maven.plugins.annotations.Mojo;
 import org.apache.maven.plugins.annotations.Parameter;
 import org.apache.maven.project.MavenProject;
+import org.codehaus.jackson.map.ObjectMapper;
+import org.codehaus.jackson.type.TypeReference;
 
+import java.io.BufferedInputStream;
+import java.io.BufferedOutputStream;
 import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.io.FileOutputStream;
+import java.io.IOException;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
-
+import java.util.Map;
+import java.util.zip.CRC32;
 
 @Mojo(name="protoc", defaultPhase = LifecyclePhase.GENERATE_SOURCES)
 public class ProtocMojo extends AbstractMojo {
@@ -49,6 +59,118 @@ public class ProtocMojo extends AbstractMojo {
   @Parameter(required=true)
   private String protocVersion;
 
+  @Parameter(defaultValue =
+      "${project.build.directory}/hadoop-maven-plugins-protoc-checksums.json")
+  private String checksumPath;
+
+  /**
+   * Compares include and source file checksums against previously computed
+   * checksums stored in a json file in the build directory.
+   */
+  public class ChecksumComparator {
+
+    private final Map<String, Long> storedChecksums;
+    private final Map<String, Long> computedChecksums;
+
+    private final File checksumFile;
+
+    ChecksumComparator(String checksumPath) throws IOException {
+      checksumFile = new File(checksumPath);
+      // Read in the checksums
+      if (checksumFile.exists()) {
+        ObjectMapper mapper = new ObjectMapper();
+        storedChecksums = mapper
+            .readValue(checksumFile, new TypeReference<Map<String, Long>>() {
+            });
+      } else {
+        storedChecksums = new HashMap<>(0);
+      }
+      computedChecksums = new HashMap<>();
+    }
+
+    public boolean hasChanged(File file) throws IOException {
+      if (!file.exists()) {
+        throw new FileNotFoundException(
+            "Specified protoc include or source does not exist: " + file);
+      }
+      if (file.isDirectory()) {
+        return hasDirectoryChanged(file);
+      } else if (file.isFile()) {
+        return hasFileChanged(file);
+      } else {
+        throw new IOException("Not a file or directory: " + file);
+      }
+    }
+
+    private boolean hasDirectoryChanged(File directory) throws IOException {
+      File[] listing = directory.listFiles();
+      boolean changed = false;
+      // Do not exit early, since we need to compute and save checksums
+      // for each file within the directory.
+      for (File f : listing) {
+        if (f.isDirectory()) {
+          if (hasDirectoryChanged(f)) {
+            changed = true;
+          }
+        } else if (f.isFile()) {
+          if (hasFileChanged(f)) {
+            changed = true;
+          }
+        } else {
+          getLog().debug("Skipping entry that is not a file or directory: "
+              + f);
+        }
+      }
+      return changed;
+    }
+
+    private boolean hasFileChanged(File file) throws IOException {
+      long computedCsum = computeChecksum(file);
+
+      // Return if the generated csum matches the stored csum
+      Long storedCsum = storedChecksums.get(file.getCanonicalPath());
+      if (storedCsum == null || storedCsum.longValue() != computedCsum) {
+        // It has changed.
+        return true;
+      }
+      return false;
+    }
+
+    private long computeChecksum(File file) throws IOException {
+      // If we've already computed the csum, reuse the computed value
+      final String canonicalPath = file.getCanonicalPath();
+      if (computedChecksums.containsKey(canonicalPath)) {
+        return computedChecksums.get(canonicalPath);
+      }
+      // Compute the csum for the file
+      CRC32 crc = new CRC32();
+      byte[] buffer = new byte[1024*64];
+      try (BufferedInputStream in =
+          new BufferedInputStream(new FileInputStream(file))) {
+        while (true) {
+          int read = in.read(buffer);
+          if (read <= 0) {
+            break;
+          }
+          crc.update(buffer, 0, read);
+        }
+      }
+      // Save it in the generated map and return
+      final long computedCsum = crc.getValue();
+      computedChecksums.put(canonicalPath, computedCsum);
+      return crc.getValue();
+    }
+
+    public void writeChecksums() throws IOException {
+      ObjectMapper mapper = new ObjectMapper();
+      try (BufferedOutputStream out = new BufferedOutputStream(
+          new FileOutputStream(checksumFile))) {
+        mapper.writeValue(out, computedChecksums);
+        getLog().info("Wrote protoc checksums to file " + checksumFile);
+      }
+    }
+  }
+
   public void execute() throws MojoExecutionException {
     try {
       List<String> command = new ArrayList<String>();
@@ -58,7 +180,7 @@ public void execute() throws MojoExecutionException {
       List<String> out = new ArrayList<String>();
       if (exec.run(command, out) == 127) {
         getLog().error("protoc, not found at: " + protocCommand);
-        throw new MojoExecutionException("protoc failure");        
+        throw new MojoExecutionException("protoc failure");
       } else {
         if (out.isEmpty()) {
           getLog().error("stdout: " + out);
@@ -67,36 +189,74 @@ public void execute() throws MojoExecutionException {
         } else {
           if (!out.get(0).endsWith(protocVersion)) {
             throw new MojoExecutionException(
-                "protoc version is '" + out.get(0) + "', expected version is '" 
+                "protoc version is '" + out.get(0) + "', expected version is '"
-                    + protocVersion + "'");            
+                    + protocVersion + "'");
           }
         }
       }
       if (!output.mkdirs()) {
         if (!output.exists()) {
-          throw new MojoExecutionException("Could not create directory: " + 
+          throw new MojoExecutionException(
-            output);
+              "Could not create directory: " + output);
         }
       }
+
+      // Whether the import or source protoc files have changed.
+      ChecksumComparator comparator = new ChecksumComparator(checksumPath);
+      boolean importsChanged = false;
+
       command = new ArrayList<String>();
       command.add(protocCommand);
       command.add("--java_out=" + output.getCanonicalPath());
       if (imports != null) {
         for (File i : imports) {
+          if (comparator.hasChanged(i)) {
+            importsChanged = true;
+          }
           command.add("-I" + i.getCanonicalPath());
         }
       }
+      // Filter to generate classes for just the changed source files.
+      List<File> changedSources = new ArrayList<>();
+      boolean sourcesChanged = false;
       for (File f : FileSetUtils.convertFileSetToFiles(source)) {
-        command.add(f.getCanonicalPath());
+        // Need to recompile if the source has changed, or if any import has
-      }
+        // changed.
-      exec = new Exec(this);
+        if (comparator.hasChanged(f) || importsChanged) {
-      out = new ArrayList<String>();
+          sourcesChanged = true;
-      if (exec.run(command, out) != 0) {
+          changedSources.add(f);
-        getLog().error("protoc compiler error");
+          command.add(f.getCanonicalPath());
-        for (String s : out) {
-          getLog().error(s);
         }
-        throw new MojoExecutionException("protoc failure");
+      }
+
+      if (!sourcesChanged && !importsChanged) {
+        getLog().info("No changes detected in protoc files, skipping "
+            + "generation.");
+      } else {
+        if (getLog().isDebugEnabled()) {
+          StringBuilder b = new StringBuilder();
+          b.append("Generating classes for the following protoc files: [");
+          String prefix = "";
+          for (File f : changedSources) {
+            b.append(prefix);
+            b.append(f.toString());
+            prefix = ", ";
+          }
+          b.append("]");
+          getLog().debug(b.toString());
+        }
+
+        exec = new Exec(this);
+        out = new ArrayList<String>();
+        if (exec.run(command, out) != 0) {
+          getLog().error("protoc compiler error");
+          for (String s : out) {
+            getLog().error(s);
+          }
+          throw new MojoExecutionException("protoc failure");
+        }
+        // Write the new checksum file on success.
+        comparator.writeChecksums();
       }
     } catch (Throwable ex) {
       throw new MojoExecutionException(ex.toString(), ex);