Initial (#1)

* Initial commit.

* Added documentation.

* Added documentation.

* Change the version to 1.0.0

Jenkinsfile

@@ -0,0 +1,5 @@
+#!groovy
+@Library('jenkins-pipeline') import com.github.jcustenborder.jenkins.pipeline.KafkaConnectPipeline
+
+def pipe = new KafkaConnectPipeline()
+pipe.execute()

README.md

@@ -1,2 +1,19 @@
-# kafka-connect-transform-archive
-Kafka Connect transform to assist with archiving to S3. 
+# Introduction
+
+Kafka Connect transform to assist with archiving to S3.
+
+## Kafka 0.11.x and newer
+
+```properties
+
+
+```
+
+## Kafka 0.10.x and older
+
+```properties
+
+
+
+```
+

docs/transformations.rst

@@ -0,0 +1,21 @@
+=================
+Archive Transform
+=================
+
+The Archive transformation is used to help preserve all of the data for a message when archived to S3.
+
+.. toctree::
+    :maxdepth: 1
+    :caption: Transformations:
+    :hidden:
+    :glob:
+
+    transformations/*
+
+
+.. toctree::
+    :maxdepth: 0
+    :caption: Schemas:
+    :hidden:
+
+    schemas

pom.xml

@@ -0,0 +1,77 @@
+<?xml version="1.0"?>
+<!--
+
+    Copyright © 2017 Jeremy Custenborder ([email protected])
+
+    Licensed under the Apache License, Version 2.0 (the "License");
+    you may not use this file except in compliance with the License.
+    You may obtain a copy of the License at
+
+        http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing, software
+    distributed under the License is distributed on an "AS IS" BASIS,
+    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+    See the License for the specific language governing permissions and
+    limitations under the License.
+
+-->
+<project xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://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>com.github.jcustenborder.kafka.connect</groupId>
+    <artifactId>kafka-connect-parent</artifactId>
+    <version>1.0.0</version>
+  </parent>
+  <artifactId>kafka-connect-transform-archive</artifactId>
+  <version>0.1.0-SNAPSHOT</version>
+  <name>kafka-connect-transform-archive</name>
+  <url>https://github.com/jcustenborder/kafka-connect-transform-archive</url>
+  <inceptionYear>2017</inceptionYear>
+  <licenses>
+    <license>
+      <name>Apache License 2.0</name>
+      <url>https://www.apache.org/licenses/LICENSE-2.0.txt</url>
+      <distribution>repo</distribution>
+    </license>
+  </licenses>
+  <developers>
+    <developer>
+      <name>Jeremy Custenborder</name>
+      <email>[email protected]</email>
+      <url>https://github.com/jcustenborder</url>
+      <roles>
+        <role>maintainer</role>
+      </roles>
+    </developer>
+  </developers>
+  <scm>
+    <connection>scm:git:https://github.com/jcustenborder/kafka-connect-transform-archive.git</connection>
+    <developerConnection>scm:git:[email protected]:jcustenborder/kafka-connect-transform-archive.git</developerConnection>
+    <url>https://github.com/jcustenborder/kafka-connect-transform-archive</url>
+  </scm>
+  <issueManagement>
+    <system>github</system>
+    <url>https://github.com/jcustenborder/kafka-connect-transform-archive/issues</url>
+  </issueManagement>
+  <dependencies>
+    <dependency>
+      <groupId>com.github.jcustenborder</groupId>
+      <artifactId>cef-parser</artifactId>
+      <version>[0.0.1.7,0.0.1.2000)</version>
+    </dependency>
+    <dependency>
+      <groupId>org.reflections</groupId>
+      <artifactId>reflections</artifactId>
+      <version>0.9.10</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>com.github.jcustenborder.kafka.connect</groupId>
+      <artifactId>connect-utils-testing-data</artifactId>
+      <version>[0.2.33,0.2.1000)</version>
+      <scope>test</scope>
+    </dependency>
+  </dependencies>
+</project>

src/main/java/com/github/jcustenborder/kafka/connect/archive/Archive.java

@@ -0,0 +1,64 @@
+/**
+ * Copyright © 2017 Jeremy Custenborder ([email protected])
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.github.jcustenborder.kafka.connect.archive;
+
+import com.github.jcustenborder.kafka.connect.utils.config.Description;
+import com.github.jcustenborder.kafka.connect.utils.config.DocumentationNote;
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.connect.connector.ConnectRecord;
+import org.apache.kafka.connect.data.Schema;
+import org.apache.kafka.connect.data.SchemaBuilder;
+import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.transforms.Transformation;
+
+import java.util.Map;
+
+@Description("The Archive transformation is used to help preserve all of the data for a message when archived to S3.")
+@DocumentationNote("This transform works by copying the key, value, topic, and timestamp to new record where this is all " +
+    "contained in the value of the message. This will allow connectors like Confluent's S3 connector to properly archive " +
+    "the record.")
+public class Archive<R extends ConnectRecord<R>> implements Transformation<R> {
+  @Override
+  public R apply(R r) {
+    final Schema schema = SchemaBuilder.struct()
+        .name("com.github.jcustenborder.kafka.connect.archive.Storage")
+        .field("key", r.keySchema())
+        .field("value", r.valueSchema())
+        .field("topic", Schema.STRING_SCHEMA)
+        .field("timestamp", Schema.INT64_SCHEMA);
+    Struct value = new Struct(schema)
+        .put("key", r.key())
+        .put("value", r.value())
+        .put("topic", r.topic())
+        .put("timestamp", r.timestamp());
+    return r.newRecord(r.topic(), r.kafkaPartition(), null, null, schema, value, r.timestamp());
+  }
+
+  @Override
+  public ConfigDef config() {
+    return new ConfigDef();
+  }
+
+  @Override
+  public void close() {
+
+  }
+
+  @Override
+  public void configure(Map<String, ?> map) {
+
+  }
+}