1 0 2 20 cdc

.gitignore

@@ -14,3 +14,12 @@ mainGeneratedRest/
 mainGeneratedDataTemplate/
 .DS_Store
 ligradle/
+
+# Kafka Web View data
+kafkawebview_data/
+
+# Local server properties
+server-local.properties
+
+# Local Brooklin data
+.brooklin/

build.gradle

@@ -154,18 +154,25 @@ project(':datastream-common') {
     compile "com.intellij:annotations:$intellijAnnotationsVersion"
     compile "com.google.guava:guava:$guavaVersion"
     compile "tech.allegro.schema.json2avro:converter:$json2avroVersion"
+    compile "com.wayfair.commons:commons-metrics:$wayfairMetricsVersion"
+
     testCompile "org.mockito:mockito-core:$mockitoVersion"
   }
 }
 
 project(':datastream-bigquery') {
+  checkstyle {
+    configFile = file("${project.rootDir}/checkstyle/checkstyle.xml")
+    sourceSets = [ getProject().sourceSets.main ]
+    toolVersion = "6.7"
+    ignoreFailures = false
+  }
   dependencies {
 
     compile "org.apache.avro:avro:$avroVersion"
     compile "org.apache.kafka:kafka-clients:$kafkaClientsVersion"
 
     compile "io.confluent:kafka-avro-serializer:$kafkaSchemaRegistryClientVersion"
-    compile "io.confluent:kafka-schema-registry:$kafkaSchemaRegistryVersion"
     compile "io.confluent:kafka-schema-registry-client:$kafkaSchemaRegistryClientVersion"
 
     compile "com.google.cloud:google-cloud-storage:$googleCloudStorageVersion"
@@ -174,6 +181,7 @@ project(':datastream-bigquery') {
     compile project(':datastream-server')
     compile project(':datastream-utils')
 
+    testCompile project(':datastream-testcommon')
     testCompile "org.mockito:mockito-core:$mockitoVersion"
 
     tasks.create(name: "copyDependentLibs", type: Copy) {
@@ -199,7 +207,6 @@ project(':datastream-cloud-storage') {
     compile "org.apache.parquet:parquet-column:$parquetWriterVersion"
 
     compile "io.confluent:kafka-avro-serializer:$kafkaSchemaRegistryClientVersion"
-    compile "io.confluent:kafka-schema-registry:$kafkaSchemaRegistryVersion"
     compile "io.confluent:kafka-schema-registry-client:$kafkaSchemaRegistryClientVersion"
 
     compile "com.google.cloud:google-cloud-storage:$googleCloudStorageVersion"
@@ -225,6 +232,7 @@ project(':datastream-server-api') {
   dependencies {
     compile project(':datastream-common')
     compile project(':datastream-utils')
+    testCompile "org.mockito:mockito-core:$mockitoVersion"
   }
 }
 
@@ -352,7 +360,7 @@ project(':datastream-tools') {
     compile "commons-cli:commons-cli:$commonsCliVersion"
   }
 
-  tasks.create(name: "releaseTarGz", dependsOn: configurations.archives.artifacts, type: Tar) {
+  tasks.create(name: "releaseTarGz", group: "distribution", dependsOn: configurations.archives.artifacts, type: Tar) {
     description = "Create Brooklin release tarball"
     baseName = "${rootProject.name}"
     into { "${property('baseName')}-$rootProject.version" }
@@ -387,6 +395,15 @@ project(':datastream-tools') {
     from(project(':datastream-kafka-connector').configurations.runtime) { into("libs/") }
     duplicatesStrategy 'exclude'
   }
+  tasks.create(name: "copyDependentLibs", type: Copy) {
+    from (configurations.runtime) {
+    }
+    into "$buildDir/dependent-libs"
+  }
+
+  jar {
+    dependsOn 'copyDependentLibs'
+  }
 }
 
 project(':datastream-client') {

config/log4j.properties

@@ -5,7 +5,9 @@ log4j.appender.stdout.layout.ConversionPattern=[%d] %p %m (%c)%n
 log4j.appender.datastreamServiceAppender=org.apache.log4j.DailyRollingFileAppender
 log4j.appender.datastreamServiceAppender.DatePattern='.'yyyy-MM-dd-HH
 log4j.appender.datastreamServiceAppender.File=${datastream.logs.dir}/server.log
+log4j.appender.file.MaxFileSize=200MB
 log4j.appender.datastreamServiceAppender.layout=org.apache.log4j.PatternLayout
-log4j.appender.datastreamServiceAppender.layout.ConversionPattern=[%d] %p %m (%c)%n
+log4j.appender.datastreamServiceAppender.layout.ConversionPattern=[%d{dd MMM yyyy HH:mm:ss,SSS}] [%p] [%t] [%c:%L] %m %n
 
-log4j.rootLogger=INFO, datastreamServiceAppender , stdout
+log4j.rootLogger=INFO, datastreamServiceAppender
+log4j.logger.com.linkedin.datastream=DEBUG

datastream-bigquery/src/main/java/com/linkedin/datastream/bigquery/Batch.java

@@ -5,30 +5,33 @@
  */
 package com.linkedin.datastream.bigquery;
 
-import java.nio.BufferUnderflowException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Optional;
 
 import org.apache.avro.generic.GenericRecord;
-import org.apache.kafka.common.errors.SerializationException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import com.google.cloud.bigquery.InsertAllRequest;
 import com.google.cloud.bigquery.Schema;
 
-import io.confluent.kafka.serializers.KafkaAvroDeserializer;
-
+import com.linkedin.datastream.bigquery.schema.BigquerySchemaEvolver;
 import com.linkedin.datastream.bigquery.translator.RecordTranslator;
 import com.linkedin.datastream.bigquery.translator.SchemaTranslator;
 import com.linkedin.datastream.common.DatastreamRecordMetadata;
 import com.linkedin.datastream.common.Package;
 import com.linkedin.datastream.metrics.DynamicMetricsManager;
+import com.linkedin.datastream.serde.Deserializer;
 import com.linkedin.datastream.server.api.transport.buffered.AbstractBatch;
 
 /**
  * This class populates a batch of BQ rows to be committed.
  */
 public class Batch extends AbstractBatch {
 
+    private static final Logger _logger = LoggerFactory.getLogger(Batch.class);
+
     private final int _maxBatchSize;
     private final int _maxBatchAge;
     private final BigqueryBatchCommitter _committer;
@@ -38,38 +41,44 @@ public class Batch extends AbstractBatch {
 
     private org.apache.avro.Schema _avroSchema;
     private Schema _schema;
-    private SchemaRegistry _schemaRegistry;
-    private String _destination;
-    private KafkaAvroDeserializer _deserializer;
+    private final BigqueryDatastreamDestination _destination;
+    private final Deserializer _valueDeserializer;
+    private final BigquerySchemaEvolver _schemaEvolver;
+    private final org.apache.avro.Schema _fixedAvroSchema;
 
     /**
      * Constructor for Batch.
      * @param maxBatchSize any batch bigger than this threshold will be committed to BQ.
      * @param maxBatchAge any batch older than this threshold will be committed to BQ.
      * @param maxInflightWriteLogCommits maximum allowed batches in the commit backlog
-     * @param schemaRegistry schema registry client object
+     * @param valueDeserializer a Deserializer
      * @param committer committer object
      */
-    public Batch(int maxBatchSize,
-                 int maxBatchAge,
-                 int maxInflightWriteLogCommits,
-                 SchemaRegistry schemaRegistry,
-                 BigqueryBatchCommitter committer) {
+    public Batch(
+            final BigqueryDatastreamDestination destination,
+            final int maxBatchSize,
+            final int maxBatchAge,
+            final int maxInflightWriteLogCommits,
+            final Deserializer valueDeserializer,
+            final BigqueryBatchCommitter committer,
+            final BigquerySchemaEvolver schemaEvolver,
+            final org.apache.avro.Schema fixedAvroSchema
+    ) {
         super(maxInflightWriteLogCommits);
+        this._destination = destination;
         this._maxBatchSize = maxBatchSize;
         this._maxBatchAge = maxBatchAge;
         this._committer = committer;
+        this._valueDeserializer = valueDeserializer;
+        this._schemaEvolver = schemaEvolver;
+        this._fixedAvroSchema = fixedAvroSchema;
         this._batch = new ArrayList<>();
         this._batchCreateTimeStamp = System.currentTimeMillis();
         this._schema = null;
-        this._destination = null;
-        this._schemaRegistry = schemaRegistry;
-        this._deserializer = _schemaRegistry.getDeserializer();
     }
 
     private void reset() {
         _batch.clear();
-        _destination = null;
         _ackCallbacks.clear();
         _recordMetadata.clear();
         _sourceTimestamps.clear();
@@ -82,7 +91,7 @@ public void write(Package aPackage) throws InterruptedException {
 
             // skip null records
             if (aPackage.getRecord().getValue() == null) {
-                LOG.info("Null record received from topic {}, partition {}, and offset {}",
+                _logger.info("Null record received from topic {}, partition {}, and offset {}",
                         aPackage.getTopic(), aPackage.getPartition(), aPackage.getOffset());
                 DynamicMetricsManager.getInstance().createOrUpdateMeter(
                         this.getClass().getSimpleName(),
@@ -94,81 +103,84 @@ public void write(Package aPackage) throws InterruptedException {
                 return;
             }
 
-            if (_destination == null) {
-                String[] datasetRetentionTableSuffix = aPackage.getDestination().split("/");
-                if (datasetRetentionTableSuffix.length == 3) {
-                    _destination = datasetRetentionTableSuffix[0] +
-                            "/" + aPackage.getTopic() + datasetRetentionTableSuffix[2] +
-                            "/" + datasetRetentionTableSuffix[1];
-                } else {
-                    _destination = datasetRetentionTableSuffix[0] +
-                            "/" + aPackage.getTopic() +
-                            "/" + datasetRetentionTableSuffix[1];
-                }
-            }
-
-            if (_schema == null) {
-                _avroSchema = _schemaRegistry.getSchemaByTopic(aPackage.getTopic());
-                _schema = SchemaTranslator.translate(_avroSchema);
-                _committer.setDestTableSchema(_destination, _schema);
-            }
-
             GenericRecord record;
             try {
-                record = (GenericRecord) _deserializer.deserialize(
+                record = (GenericRecord) _valueDeserializer.deserialize((byte[]) aPackage.getRecord().getValue());
+            } catch (Exception e) {
+                _logger.warn("Error deserializing message at Topic {} - Partition {} - Offset {} - Reason {} - Exception {} for destination {}",
                         aPackage.getTopic(),
-                        (byte[]) aPackage.getRecord().getValue());
-            } catch (SerializationException e) {
-                if (e.getCause() instanceof BufferUnderflowException) {
-                    LOG.warn("Skipping message at Topic {} - Partition {} - Offset {} - Reason {} - Exception {}",
-                            aPackage.getTopic(),
-                            aPackage.getPartition(),
-                            aPackage.getOffset(),
-                            e.getMessage(),
-                            e.getCause().getClass());
-                    DynamicMetricsManager.getInstance().createOrUpdateMeter(
-                            this.getClass().getSimpleName(),
-                            aPackage.getTopic(),
-                            "deserializerErrorCount",
-                            1);
-                    aPackage.getAckCallback().onCompletion(new DatastreamRecordMetadata(
-                            aPackage.getCheckpoint(), aPackage.getTopic(), aPackage.getPartition()), null);
-                    return;
-                } else {
-                    throw e;
-                }
+                        aPackage.getPartition(),
+                        aPackage.getOffset(),
+                        e.getMessage(),
+                        e.getCause().getClass(),
+                        _destination,
+                        e);
+                DynamicMetricsManager.getInstance().createOrUpdateMeter(
+                        this.getClass().getSimpleName(),
+                        aPackage.getTopic(),
+                        "deserializerErrorCount",
+                        1);
+                throw e;
             }
 
-            _batch.add(RecordTranslator.translate(record, _avroSchema));
+            if (_fixedAvroSchema == null) {
+                processAvroSchema(record.getSchema());
+            } else {
+                processAvroSchema(_fixedAvroSchema);
+            }
+
+            _batch.add(RecordTranslator.translate(record));
 
             _ackCallbacks.add(aPackage.getAckCallback());
             _recordMetadata.add(new DatastreamRecordMetadata(aPackage.getCheckpoint(),
                     aPackage.getTopic(),
                     aPackage.getPartition()));
             _sourceTimestamps.add(aPackage.getTimestamp());
-        } else if (aPackage.isTryFlushSignal() || aPackage.isForceFlushSignal()) {
-            if (_batch.isEmpty()) {
-                LOG.debug("Nothing to flush.");
-                return;
+        }
+
+        if (aPackage.isForceFlushSignal() || _batch.size() >= _maxBatchSize ||
+                System.currentTimeMillis() - _batchCreateTimeStamp >= _maxBatchAge) {
+            flush(aPackage.isForceFlushSignal());
+        }
+    }
+
+    private void processAvroSchema(final org.apache.avro.Schema avroSchema) {
+        final Optional<Schema> newBQSchema;
+        if (_avroSchema == null) {
+            newBQSchema = Optional.of(SchemaTranslator.translate(avroSchema));
+        } else if (!_avroSchema.equals(avroSchema)) {
+            final Schema evolvedSchema = _schemaEvolver.evolveSchema(_schema, SchemaTranslator.translate(avroSchema));
+            if (!_schema.equals(evolvedSchema)) {
+                newBQSchema = Optional.of(evolvedSchema);
+            } else {
+                newBQSchema = Optional.empty();
             }
+        } else {
+            newBQSchema = Optional.empty();
         }
+        newBQSchema.ifPresent(schema -> {
+            _avroSchema = avroSchema;
+            _schema = schema;
+            _committer.setDestTableSchema(_destination, _schema);
+        });
+    }
 
-        if (_batch.size() >= _maxBatchSize ||
-                System.currentTimeMillis() - _batchCreateTimeStamp >= _maxBatchAge ||
-                aPackage.isForceFlushSignal()) {
+    private void flush(final boolean force) throws InterruptedException {
+        if (_batch.isEmpty()) {
+            _logger.debug("Nothing to flush.");
+        } else {
             waitForRoomInCommitBacklog();
             incrementInflightWriteLogCommits();
             _committer.commit(
                     new ArrayList<>(_batch),
-                    _destination,
+                    _destination.toString(),
                     new ArrayList<>(_ackCallbacks),
                     new ArrayList<>(_recordMetadata),
                     new ArrayList<>(_sourceTimestamps),
-                    () -> decrementInflightWriteLogCommitsAndNotify()
+                    this::decrementInflightWriteLogCommitsAndNotify
             );
             reset();
-
-            if (aPackage.isForceFlushSignal()) {
+            if (force) {
                 waitForCommitBacklogToClear();
             }
         }