Merge pull request #42 from GoogleCloudPlatform/gcs-s3-inspect

Fix DLP client version conflict with beam 2.16 and change window strategy for S3 scan

Author: santhh

build.gradle

@@ -70,10 +70,9 @@ dependencies {
     compile group: 'org.apache.beam', name: 'beam-runners-direct-java', version: dataflowBeamVersion
 	compile group: 'org.slf4j', name: 'slf4j-jdk14', version: '1.7.5'
 	compile 'com.google.cloud:google-cloud-kms:0.70.0-beta'   
-    compile 'com.google.cloud:google-cloud-dlp:0.74.0-beta'
     compile 'com.google.guava:guava:27.0-jre'
+	compile group: 'com.google.cloud', name: 'google-cloud-dlp', version: '0.100.0-beta'
     compile 'com.google.api-client:google-api-client:1.27.0'
-   	//compile group: 'com.amazonaws', name: 'aws-java-sdk', version: '1.11.546'
     compile group: 'com.google.apis', name: 'google-api-services-cloudkms', version: 'v1-rev53-1.23.0'
 	compile group: 'org.apache.beam', name: 'beam-sdks-java-io-amazon-web-services', version: dataflowBeamVersion
 	compile "com.google.auto.value:auto-value-annotations:1.6.2"

src/main/java/com/google/swarm/tokenization/S3Import.java

@@ -59,8 +59,6 @@
 import org.apache.beam.sdk.transforms.WithKeys;
 import org.apache.beam.sdk.transforms.splittabledofn.OffsetRangeTracker;
 import org.apache.beam.sdk.transforms.splittabledofn.RestrictionTracker;
-import org.apache.beam.sdk.transforms.windowing.AfterPane;
-import org.apache.beam.sdk.transforms.windowing.AfterProcessingTime;
 import org.apache.beam.sdk.transforms.windowing.AfterWatermark;
 import org.apache.beam.sdk.transforms.windowing.FixedWindows;
 import org.apache.beam.sdk.transforms.windowing.Window;
@@ -85,8 +83,8 @@ public class S3Import {
   private static Integer BATCH_SIZE = 520000;
   private static Integer DLP_PAYLOAD_LIMIT = 524288;
   private static final String BQ_TABLE_NAME = String.valueOf("S3_DLP_INSPECT_FINDINGS");
-  private static final Duration DEFAULT_POLL_INTERVAL = Duration.standardSeconds(30);
-  private static final Duration WINDOW_INTERVAL = Duration.standardSeconds(5);
+  private static final Duration DEFAULT_POLL_INTERVAL = Duration.standardSeconds(60);
+  private static final Duration WINDOW_INTERVAL = Duration.standardSeconds(10);
   private static final DateTimeFormatter TIMESTAMP_FORMATTER =
       DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss.SSSSSS");
 
@@ -122,12 +120,7 @@ public static void main(String[] args) {
         s3Files.apply(
             "Fixed Window",
             Window.<KV<String, ReadableFile>>into(FixedWindows.of(WINDOW_INTERVAL))
-                .triggering(
-                    AfterWatermark.pastEndOfWindow()
-                        .withEarlyFirings(
-                            AfterProcessingTime.pastFirstElementInPane()
-                                .plusDelayOf(Duration.standardSeconds(10)))
-                        .withLateFirings(AfterPane.elementCountAtLeast(1)))
+                .triggering(AfterWatermark.pastEndOfWindow())
                 .discardingFiredPanes()
                 .withAllowedLateness(Duration.ZERO));
 
@@ -275,7 +268,6 @@ public void processElement(ProcessContext c) throws IOException {
 
       try (DlpServiceClient dlpServiceClient = DlpServiceClient.create()) {
         if (!c.element().getValue().isEmpty()) {
-
           ContentItem contentItem =
               ContentItem.newBuilder().setValue(c.element().getValue()).build();
           this.requestBuilder.setItem(contentItem);