Merge remote-tracking branch 'upstream/master'

JGit/pom.xml

@@ -3,10 +3,11 @@
     xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.baeldung</groupId>
-    <artifactId>JGitSnippets</artifactId>
+    <artifactId>JGit</artifactId>
     <version>1.0-SNAPSHOT</version>
     <packaging>jar</packaging>
     <url>http://maven.apache.org</url>
+	<name>JGit</name>
 
     <parent>
         <groupId>com.baeldung</groupId>

algorithms/src/main/java/com/baeldung/algorithms/distancebetweenpoints/DistanceBetweenPointsService.java

@@ -0,0 +1,38 @@
+package com.baeldung.algorithms.distancebetweenpoints;
+
+import java.awt.geom.Point2D;
+
+public class DistanceBetweenPointsService {
+
+    public double calculateDistanceBetweenPoints(
+        double x1, 
+        double y1, 
+        double x2, 
+        double y2) {
+        
+        return Math.sqrt((y2 - y1) * (y2 - y1) + (x2 - x1) * (x2 - x1));
+    }
+
+    public double calculateDistanceBetweenPointsWithHypot(
+        double x1, 
+        double y1, 
+        double x2, 
+        double y2) {
+        
+        double ac = Math.abs(y2 - y1);
+        double cb = Math.abs(x2 - x1);
+        
+        return Math.hypot(ac, cb);
+    }
+
+    public double calculateDistanceBetweenPointsWithPoint2D( 
+        double x1, 
+        double y1, 
+        double x2, 
+        double y2) {
+        
+        return Point2D.distance(x1, y1, x2, y2);
+
+    }
+
+}

algorithms/src/test/java/com/baeldung/algorithms/distancebetweenpoints/DistanceBetweenPointsServiceUnitTest.java

@@ -0,0 +1,54 @@
+package com.baeldung.algorithms.distancebetweenpoints;
+
+import org.junit.Test;
+
+import com.baeldung.algorithms.distancebetweenpoints.DistanceBetweenPointsService;
+
+import static org.junit.Assert.assertEquals;
+
+public class DistanceBetweenPointsServiceUnitTest {
+
+    private DistanceBetweenPointsService service = new DistanceBetweenPointsService();
+
+    @Test
+    public void givenTwoPoints_whenCalculateDistanceByFormula_thenCorrect() {
+
+        double x1 = 3;
+        double y1 = 4;
+        double x2 = 7;
+        double y2 = 1;
+
+        double distance = service.calculateDistanceBetweenPoints(x1, y1, x2, y2);
+
+        assertEquals(distance, 5, 0.001);
+
+    }
+
+    @Test
+    public void givenTwoPoints_whenCalculateDistanceWithHypot_thenCorrect() {
+
+        double x1 = 3;
+        double y1 = 4;
+        double x2 = 7;
+        double y2 = 1;
+
+        double distance = service.calculateDistanceBetweenPointsWithHypot(x1, y1, x2, y2);
+
+        assertEquals(distance, 5, 0.001);
+
+    }
+
+    @Test
+    public void givenTwoPoints_whenCalculateDistanceWithPoint2D_thenCorrect() {
+
+        double x1 = 3;
+        double y1 = 4;
+        double x2 = 7;
+        double y2 = 1;
+
+        double distance = service.calculateDistanceBetweenPointsWithPoint2D(x1, y1, x2, y2);
+
+        assertEquals(distance, 5, 0.001);
+
+    }
+}

apache-avro/pom.xml

@@ -4,8 +4,9 @@
          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.baeldung</groupId>
-    <artifactId>apache-avro-tutorial</artifactId>
+    <artifactId>apache-avro</artifactId>
     <version>0.0.1-SNAPSHOT</version>
+	<name>Apache Avro</version>
 
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>

core-java/src/main/java/com/baeldung/synthetic/BridgeMethodDemo.java

@@ -0,0 +1,23 @@
+package com.baeldung.synthetic;
+
+import java.util.Comparator;
+
+/**
+ * Class which contains a synthetic bridge method.
+ * 
+ * @author Donato Rimenti
+ *
+ */
+public class BridgeMethodDemo implements Comparator<Integer> {
+
+    /*
+     * (non-Javadoc)
+     * 
+     * @see java.util.Comparator#compare(java.lang.Object, java.lang.Object)
+     */
+    @Override
+    public int compare(Integer o1, Integer o2) {
+        return 0;
+    }
+
+}

core-java/src/main/java/com/baeldung/synthetic/SyntheticConstructorDemo.java

@@ -0,0 +1,34 @@
+package com.baeldung.synthetic;
+
+/**
+ * Wrapper for a class which contains a synthetic constructor.
+ * 
+ * @author Donato Rimenti
+ *
+ */
+public class SyntheticConstructorDemo {
+
+    /**
+     * We need to instantiate the {@link NestedClass} using a private
+     * constructor from the enclosing instance in order to generate a synthetic
+     * constructor.
+     */
+    private NestedClass nestedClass = new NestedClass();
+
+    /**
+     * Class which contains a synthetic constructor.
+     * 
+     * @author Donato Rimenti
+     *
+     */
+    class NestedClass {
+
+        /**
+         * In order to generate a synthetic constructor, this class must have a
+         * private constructor.
+         */
+        private NestedClass() {
+        }
+    }
+
+}

core-java/src/main/java/com/baeldung/synthetic/SyntheticFieldDemo.java

@@ -0,0 +1,22 @@
+package com.baeldung.synthetic;
+
+/**
+ * Wrapper for a class which contains a synthetic field reference to the outer
+ * class.
+ * 
+ * @author Donato Rimenti
+ *
+ */
+public class SyntheticFieldDemo {
+
+    /**
+     * Class which contains a synthetic field reference to the outer class.
+     * 
+     * @author Donato Rimenti
+     *
+     */
+    class NestedClass {
+
+    }
+
+}

core-java/src/main/java/com/baeldung/synthetic/SyntheticMethodDemo.java

@@ -0,0 +1,48 @@
+package com.baeldung.synthetic;
+
+/**
+ * Wrapper for a class which contains two synthetic methods accessors to a
+ * private field.
+ * 
+ * @author Donato Rimenti
+ *
+ */
+public class SyntheticMethodDemo {
+
+    /**
+     * Class which contains two synthetic methods accessors to a private field.
+     * 
+     * @author Donato Rimenti
+     *
+     */
+    class NestedClass {
+
+        /**
+         * Field for which will be generated synthetic methods accessors. It's
+         * important that this field is private for this purpose.
+         */
+        private String nestedField;
+    }
+
+    /**
+     * Gets the private nested field. We need to read the nested field in order
+     * to generate the synthetic getter.
+     * 
+     * @return the {@link NestedClass#nestedField}
+     */
+    public String getNestedField() {
+        return new NestedClass().nestedField;
+    }
+
+    /**
+     * Sets the private nested field. We need to write the nested field in order
+     * to generate the synthetic setter.
+     * 
+     * @param nestedField
+     *            the {@link NestedClass#nestedField}
+     */
+    public void setNestedField(String nestedField) {
+        new NestedClass().nestedField = nestedField;
+    }
+
+}

core-java/src/test/java/com/baeldung/synthetic/SyntheticUnitTest.java

@@ -0,0 +1,99 @@
+package com.baeldung.synthetic;
+
+import java.lang.reflect.Constructor;
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * Unit test for {@link SyntheticFieldDemo}, {@link SyntheticMethodDemo},
+ * {@link SyntheticConstructorDemo} and {@link BridgeMethodDemo} classes.
+ * 
+ * @author Donato Rimenti
+ *
+ */
+public class SyntheticUnitTest {
+
+    /**
+     * Tests that the {@link SyntheticMethodDemo.NestedClass} contains two synthetic
+     * methods.
+     */
+    @Test
+    public void givenSyntheticMethod_whenIsSinthetic_thenTrue() {
+        // Checks that the nested class contains exactly two synthetic methods.
+        Method[] methods = SyntheticMethodDemo.NestedClass.class.getDeclaredMethods();
+        Assert.assertEquals("This class should contain only two methods", 2, methods.length);
+
+        for (Method m : methods) {
+            System.out.println("Method: " + m.getName() + ", isSynthetic: " + m.isSynthetic());
+            Assert.assertTrue("All the methods of this class should be synthetic", m.isSynthetic());
+        }
+    }
+
+    /**
+     * Tests that {@link SyntheticConstructorDemo.NestedClass} contains a synthetic
+     * constructor.
+     */
+    @Test
+    public void givenSyntheticConstructor_whenIsSinthetic_thenTrue() {
+        // Checks that the nested class contains exactly a synthetic
+        // constructor.
+        int syntheticConstructors = 0;
+        Constructor<?>[] constructors = SyntheticConstructorDemo.NestedClass.class.getDeclaredConstructors();
+        Assert.assertEquals("This class should contain only two constructors", 2, constructors.length);
+
+        for (Constructor<?> c : constructors) {
+            System.out.println("Constructor: " + c.getName() + ", isSynthetic: " + c.isSynthetic());
+
+            // Counts the synthetic constructors.
+            if (c.isSynthetic()) {
+                syntheticConstructors++;
+            }
+        }
+
+        // Checks that there's exactly one synthetic constructor.
+        Assert.assertEquals(1, syntheticConstructors);
+    }
+
+    /**
+     * Tests that {@link SyntheticFieldDemo.NestedClass} contains a synthetic field.
+     */
+    @Test
+    public void givenSyntheticField_whenIsSinthetic_thenTrue() {
+        // This class should contain exactly one synthetic field.
+        Field[] fields = SyntheticFieldDemo.NestedClass.class.getDeclaredFields();
+        Assert.assertEquals("This class should contain only one field", 1, fields.length);
+
+        for (Field f : fields) {
+            System.out.println("Field: " + f.getName() + ", isSynthetic: " + f.isSynthetic());
+            Assert.assertTrue("All the fields of this class should be synthetic", f.isSynthetic());
+        }
+    }
+
+    /**
+     * Tests that {@link BridgeMethodDemo} contains a synthetic bridge method.
+     */
+    @Test
+    public void givenBridgeMethod_whenIsBridge_thenTrue() {
+        // This class should contain exactly one synthetic bridge method.
+        int syntheticMethods = 0;
+        Method[] methods = BridgeMethodDemo.class.getDeclaredMethods();
+        for (Method m : methods) {
+            System.out.println(
+                    "Method: " + m.getName() + ", isSynthetic: " + m.isSynthetic() + ", isBridge: " + m.isBridge());
+
+            // Counts the synthetic methods and checks that they are also bridge
+            // methods.
+            if (m.isSynthetic()) {
+                syntheticMethods++;
+                Assert.assertTrue("The synthetic method in this class should also be a bridge method", m.isBridge());
+            }
+        }
+        
+        // Checks that there's exactly one synthetic bridge method.
+        Assert.assertEquals("There should be exactly 1 synthetic bridge method in this class", 1, syntheticMethods);
+    }
+
+}

couchbase/pom.xml

@@ -3,11 +3,11 @@
     xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.baeldung</groupId>
-    <artifactId>couchbase-sdk</artifactId>
+    <artifactId>couchbase</artifactId>
     <version>0.1-SNAPSHOT</version>
     <packaging>jar</packaging>
     <name>couchbase</name>
-    <description>Couchbase SDK Tutorials</description>
+    <description>Couchbase Tutorials</description>
 
     <parent>
         <groupId>com.baeldung</groupId>

ejb/wildfly/pom.xml

@@ -2,9 +2,10 @@
          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.baeldung.wildfly</groupId>
-    <artifactId>wildfly-example</artifactId>
+    <artifactId>wildfly</artifactId>
     <version>0.0.1-SNAPSHOT</version>
     <packaging>pom</packaging>
+	<name>wildfly</name>
 
     <parent>
         <groupId>com.baeldung.ejb</groupId>

ejb/wildfly/wildfly-mdb/pom.xml

@@ -1,8 +1,9 @@
 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <modelVersion>4.0.0</modelVersion>
-    <artifactId>widlfly-mdb</artifactId>
-
+    <artifactId>wildfly-mdb</artifactId>
+	<name>wildfly-mdb</name>
+	
     <parent>
         <groupId>com.baeldung.wildfly</groupId>
         <artifactId>wildfly-example</artifactId>

feign/pom.xml

@@ -3,7 +3,8 @@
     xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.baeldung.feign</groupId>
-    <artifactId>feign-client</artifactId>
+    <artifactId>feign</artifactId>
+	<name>feign</name>
 
     <parent>
         <groupId>com.baeldung</groupId>

google-web-toolkit/pom.xml

@@ -6,10 +6,11 @@
 	<!-- POM file generated with GWT webAppCreator -->
 	<modelVersion>4.0.0</modelVersion>
 	<groupId>com.baeldung</groupId>
-	<artifactId>google_web_toolkit</artifactId>
+	<artifactId>google-web-toolkit</artifactId>
 	<packaging>war</packaging>
 	<version>1.0-SNAPSHOT</version>
-
+	<name>google-web-toolkit</name>
+	
     <parent>
         <groupId>com.baeldung</groupId>
         <artifactId>parent-modules</artifactId>

grpc/pom.xml

@@ -1,11 +1,11 @@
 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
     xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <modelVersion>4.0.0</modelVersion>
-    <groupId>grpc</groupId>
-    <artifactId>grpc-demo</artifactId>
+    <artifactId>grpc</artifactId>
     <version>0.0.1-SNAPSHOT</version>
     <packaging>jar</packaging>
-
+	<name>grpc</name>
+	
     <parent>
         <groupId>com.baeldung</groupId>
         <artifactId>parent-modules</artifactId>

guest/remote-debugging/pom.xml

@@ -3,9 +3,10 @@
     xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.stackify</groupId>
-    <artifactId>java-remote-debugging</artifactId>
+    <artifactId>remote-debugging</artifactId>
     <version>0.0.1-SNAPSHOT</version>
     <packaging>war</packaging>
+	<name>remote-debugging</name>
 
     <parent>
         <groupId>org.springframework.boot</groupId>

jhipster/jhipster-microservice/car-app/pom.xml

@@ -10,10 +10,10 @@
         <relativePath>../../../parent-boot-1</relativePath>
     </parent>
     <groupId>com.car.app</groupId>
-    <artifactId>carapp</artifactId>
+    <artifactId>car-app</artifactId>
     <version>0.0.1-SNAPSHOT</version>
     <packaging>war</packaging>
-    <name>Carapp</name>
+    <name>car-app</name>
 
     <prerequisites>
         <maven>${maven.version}</maven>

jhipster/jhipster-microservice/dealer-app/pom.xml

@@ -3,10 +3,10 @@
     xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.dealer.app</groupId>
-    <artifactId>dealerapp</artifactId>
+    <artifactId>dealer-app</artifactId>
     <version>0.0.1-SNAPSHOT</version>
     <packaging>war</packaging>
-    <name>Dealerapp</name>
+    <name>dealer-app</name>
 
     <parent>
         <artifactId>parent-boot-1</artifactId>

jhipster/jhipster-microservice/gateway-app/pom.xml

@@ -3,10 +3,10 @@
     xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.gateway</groupId>
-    <artifactId>gateway</artifactId>
+    <artifactId>gateway-app</artifactId>
     <version>0.0.1-SNAPSHOT</version>
     <packaging>war</packaging>
-    <name>Gateway</name>
+    <name>gateway-app</name>
 
     <parent>
         <artifactId>parent-boot-1</artifactId>

jjwt/pom.xml

@@ -3,10 +3,10 @@
     xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <modelVersion>4.0.0</modelVersion>
     <groupId>io.jsonwebtoken</groupId>
-    <artifactId>jjwtfun</artifactId>
+    <artifactId>jjwt</artifactId>
     <version>0.0.1-SNAPSHOT</version>
     <packaging>jar</packaging>
-    <name>jjwtfun</name>
+    <name>jjwt</name>
     <description>Exercising the JJWT</description>
 
     <parent>

libraries/pom.xml

@@ -154,6 +154,17 @@
             <artifactId>commons-dbutils</artifactId>
             <version>${commons.dbutils.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-connector-kafka-0.11_2.11</artifactId>
+            <version>${flink.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-streaming-java_2.11</artifactId>
+            <version>${flink.version}</version>
+        </dependency>
+
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-core</artifactId>
@@ -178,7 +189,7 @@
         </dependency>
         <dependency>
             <groupId>org.apache.flink</groupId>
-            <artifactId>flink-test-utils_2.10</artifactId>
+            <artifactId>flink-test-utils_2.11</artifactId>
             <version>${flink.version}</version>
             <scope>test</scope>
         </dependency>
@@ -228,6 +239,11 @@
             <artifactId>jackson-databind</artifactId>
             <version>${jackson.version}</version>
         </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.datatype</groupId>
+            <artifactId>jackson-datatype-jsr310</artifactId>
+            <version>${jackson.version}</version>
+        </dependency>
         <!-- JDO -->
         <dependency>
             <groupId>org.datanucleus</groupId>
@@ -902,7 +918,7 @@
 
         <httpclient.version>4.5.3</httpclient.version>
         <commons.io.version>2.5</commons.io.version>
-        <flink.version>1.2.0</flink.version>
+        <flink.version>1.5.0</flink.version>
         <jackson.version>2.8.5</jackson.version>
         <neuroph.version>2.92</neuroph.version>
         <serenity.version>1.9.26</serenity.version>

libraries/src/main/java/com/baeldung/flink/FlinkDataPipeline.java

@@ -0,0 +1,82 @@
+package com.baeldung.flink;
+
+
+import com.baeldung.flink.model.Backup;
+import com.baeldung.flink.model.InputMessage;
+import com.baeldung.flink.operator.BackupAggregator;
+import com.baeldung.flink.operator.InputMessageTimestampAssigner;
+import com.baeldung.flink.operator.WordsCapitalizer;
+import org.apache.flink.streaming.api.TimeCharacteristic;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.windowing.time.Time;
+import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;
+import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011;
+
+import static com.baeldung.flink.connector.Consumers.*;
+import static com.baeldung.flink.connector.Producers.*;
+
+public class FlinkDataPipeline {
+
+    public static void capitalize() throws Exception {
+        String inputTopic = "flink_input";
+        String outputTopic = "flink_output";
+        String consumerGroup = "baeldung";
+        String address = "localhost:9092";
+
+        StreamExecutionEnvironment environment =
+          StreamExecutionEnvironment.getExecutionEnvironment();
+
+        FlinkKafkaConsumer011<String> flinkKafkaConsumer =
+          createStringConsumerForTopic(inputTopic, address, consumerGroup);
+        flinkKafkaConsumer.setStartFromEarliest();
+
+        DataStream<String> stringInputStream =
+          environment.addSource(flinkKafkaConsumer);
+
+        FlinkKafkaProducer011<String> flinkKafkaProducer =
+        createStringProducer(outputTopic, address);
+
+        stringInputStream
+          .map(new WordsCapitalizer())
+          .addSink(flinkKafkaProducer);
+
+        environment.execute();
+    }
+
+public static void createBackup () throws Exception {
+    String inputTopic = "flink_input";
+    String outputTopic = "flink_output";
+    String consumerGroup = "baeldung";
+    String kafkaAddress = "localhost:9092";
+
+    StreamExecutionEnvironment environment =
+      StreamExecutionEnvironment.getExecutionEnvironment();
+
+    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
+
+    FlinkKafkaConsumer011<InputMessage> flinkKafkaConsumer =
+      createInputMessageConsumer(inputTopic, kafkaAddress, consumerGroup);
+    flinkKafkaConsumer.setStartFromEarliest();
+
+    flinkKafkaConsumer
+      .assignTimestampsAndWatermarks(new InputMessageTimestampAssigner());
+    FlinkKafkaProducer011<Backup> flinkKafkaProducer =
+      createBackupProducer(outputTopic, kafkaAddress);
+
+    DataStream<InputMessage> inputMessagesStream =
+      environment.addSource(flinkKafkaConsumer);
+
+    inputMessagesStream
+      .timeWindowAll(Time.hours(24))
+      .aggregate(new BackupAggregator())
+      .addSink(flinkKafkaProducer);
+
+    environment.execute();
+}
+
+    public static void main(String[] args) throws Exception {
+        createBackup();
+    }
+
+}

libraries/src/main/java/com/baeldung/flink/connector/Consumers.java

@@ -0,0 +1,32 @@
+package com.baeldung.flink.connector;
+
+import com.baeldung.flink.model.InputMessage;
+import com.baeldung.flink.schema.InputMessageDeserializationSchema;
+import org.apache.flink.api.common.serialization.SimpleStringSchema;
+import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;
+
+import java.util.Properties;
+
+public class Consumers {
+
+public static FlinkKafkaConsumer011<String> createStringConsumerForTopic(
+  String topic, String kafkaAddress, String kafkaGroup ) {
+    Properties props = new Properties();
+    props.setProperty("bootstrap.servers", kafkaAddress);
+    props.setProperty("group.id",kafkaGroup);
+    FlinkKafkaConsumer011<String> consumer =
+      new FlinkKafkaConsumer011<>(topic, new SimpleStringSchema(),props);
+
+    return consumer;
+}
+
+    public static FlinkKafkaConsumer011<InputMessage> createInputMessageConsumer(String topic, String kafkaAddress, String kafkaGroup ) {
+        Properties properties = new Properties();
+        properties.setProperty("bootstrap.servers", kafkaAddress);
+        properties.setProperty("group.id",kafkaGroup);
+        FlinkKafkaConsumer011<InputMessage> consumer = new FlinkKafkaConsumer011<InputMessage>(
+                topic, new InputMessageDeserializationSchema(),properties);
+
+        return consumer;
+    }
+}

libraries/src/main/java/com/baeldung/flink/connector/Producers.java

@@ -0,0 +1,17 @@
+package com.baeldung.flink.connector;
+
+import com.baeldung.flink.model.Backup;
+import com.baeldung.flink.schema.BackupSerializationSchema;
+import org.apache.flink.api.common.serialization.SimpleStringSchema;
+import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011;
+
+public class Producers {
+
+    public static FlinkKafkaProducer011<String> createStringProducer(String topic, String kafkaAddress) {
+        return new FlinkKafkaProducer011<>(kafkaAddress, topic, new SimpleStringSchema());
+    }
+
+    public static FlinkKafkaProducer011<Backup> createBackupProducer(String topic, String kafkaAddress) {
+        return new FlinkKafkaProducer011<Backup>(kafkaAddress, topic, new BackupSerializationSchema());
+    }
+}

libraries/src/main/java/com/baeldung/flink/model/Backup.java

@@ -0,0 +1,27 @@
+package com.baeldung.flink.model;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.UUID;
+
+public class Backup {
+
+    @JsonProperty("inputMessages")
+    List<InputMessage> inputMessages;
+    @JsonProperty("backupTimestamp")
+    LocalDateTime backupTimestamp;
+    @JsonProperty("uuid")
+    UUID uuid;
+
+    public Backup(List<InputMessage> inputMessages, LocalDateTime backupTimestamp) {
+        this.inputMessages = inputMessages;
+        this.backupTimestamp = backupTimestamp;
+        this.uuid = UUID.randomUUID();
+    }
+
+    public List<InputMessage> getInputMessages() {
+        return inputMessages;
+    }
+}

libraries/src/main/java/com/baeldung/flink/model/InputMessage.java

@@ -0,0 +1,72 @@
+package com.baeldung.flink.model;
+
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import com.google.common.base.Objects;
+
+import java.time.LocalDateTime;
+
+@JsonSerialize
+public class InputMessage {
+    String sender;
+    String recipient;
+    LocalDateTime sentAt;
+    String message;
+
+    public InputMessage() {
+    }
+
+    public String getSender() {
+        return sender;
+    }
+
+    public void setSender(String sender) {
+        this.sender = sender;
+    }
+
+    public String getRecipient() {
+        return recipient;
+    }
+
+    public void setRecipient(String recipient) {
+        this.recipient = recipient;
+    }
+
+    public LocalDateTime getSentAt() {
+        return sentAt;
+    }
+
+    public void setSentAt(LocalDateTime sentAt) {
+        this.sentAt = sentAt;
+    }
+
+    public String getMessage() {
+        return message;
+    }
+
+    public void setMessage(String message) {
+        this.message = message;
+    }
+
+    public InputMessage(String sender, String recipient, LocalDateTime sentAt, String message) {
+        this.sender = sender;
+        this.recipient = recipient;
+        this.sentAt = sentAt;
+        this.message = message;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        InputMessage message1 = (InputMessage) o;
+        return Objects.equal(sender, message1.sender) &&
+                Objects.equal(recipient, message1.recipient) &&
+                Objects.equal(sentAt, message1.sentAt) &&
+                Objects.equal(message, message1.message);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hashCode(sender, recipient, sentAt, message);
+    }
+}

libraries/src/main/java/com/baeldung/flink/operator/BackupAggregator.java

@@ -0,0 +1,34 @@
+package com.baeldung.flink.operator;
+
+import com.baeldung.flink.model.Backup;
+import com.baeldung.flink.model.InputMessage;
+import org.apache.flink.api.common.functions.AggregateFunction;
+
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.List;
+
+    public class BackupAggregator implements AggregateFunction<InputMessage, List<InputMessage>, Backup> {
+        @Override
+        public List<InputMessage> createAccumulator() {
+            return new ArrayList<>();
+        }
+
+        @Override
+        public List<InputMessage> add(InputMessage inputMessage, List<InputMessage> inputMessages) {
+            inputMessages.add(inputMessage);
+            return inputMessages;
+        }
+
+        @Override
+        public Backup getResult(List<InputMessage> inputMessages) {
+            Backup backup = new Backup(inputMessages, LocalDateTime.now());
+                return backup;
+        }
+
+        @Override
+        public List<InputMessage> merge(List<InputMessage> inputMessages, List<InputMessage> acc1) {
+            inputMessages.addAll(acc1);
+            return inputMessages;
+        }
+    }

libraries/src/main/java/com/baeldung/flink/operator/InputMessageTimestampAssigner.java

@@ -0,0 +1,23 @@
+package com.baeldung.flink.operator;
+
+import com.baeldung.flink.model.InputMessage;
+import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks;
+import org.apache.flink.streaming.api.watermark.Watermark;
+
+import javax.annotation.Nullable;
+import java.time.ZoneId;
+
+public class InputMessageTimestampAssigner implements AssignerWithPunctuatedWatermarks<InputMessage> {
+
+    @Override
+    public long extractTimestamp(InputMessage element, long previousElementTimestamp) {
+        ZoneId zoneId = ZoneId.systemDefault();
+        return element.getSentAt().atZone(zoneId).toEpochSecond() * 1000;
+    }
+
+    @Nullable
+    @Override
+    public Watermark checkAndGetNextWatermark(InputMessage lastElement, long extractedTimestamp) {
+        return new Watermark(extractedTimestamp - 15);
+    }
+}

libraries/src/main/java/com/baeldung/flink/operator/WordsCapitalizer.java

@@ -0,0 +1,11 @@
+package com.baeldung.flink.operator;
+
+import org.apache.flink.api.common.functions.MapFunction;
+
+public class WordsCapitalizer implements MapFunction<String, String> {
+
+    @Override
+    public String map(String s) {
+        return s.toUpperCase();
+    }
+}

libraries/src/main/java/com/baeldung/flink/schema/BackupSerializationSchema.java

@@ -0,0 +1,33 @@
+package com.baeldung.flink.schema;
+
+import com.baeldung.flink.model.Backup;
+import com.fasterxml.jackson.annotation.JsonAutoDetect;
+import com.fasterxml.jackson.annotation.PropertyAccessor;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class BackupSerializationSchema
+  implements SerializationSchema<Backup> {
+
+    static ObjectMapper objectMapper = new ObjectMapper().registerModule(new JavaTimeModule());
+
+    Logger logger = LoggerFactory.getLogger(BackupSerializationSchema.class);
+
+    @Override
+    public byte[] serialize(Backup backupMessage) {
+        if(objectMapper == null) {
+            objectMapper.setVisibility(PropertyAccessor.FIELD, JsonAutoDetect.Visibility.ANY);
+            objectMapper = new ObjectMapper().registerModule(new JavaTimeModule());
+        }
+        try {
+            String json = objectMapper.writeValueAsString(backupMessage);
+            return json.getBytes();
+        } catch (com.fasterxml.jackson.core.JsonProcessingException e) {
+            logger.error("Failed to parse JSON", e);
+        }
+        return new byte[0];
+    }
+}