Merge remote-tracking branch 'eugenp/master'

JGit/pom.xml

@@ -5,9 +5,9 @@
     <groupId>com.baeldung</groupId>
     <artifactId>JGit</artifactId>
     <version>1.0-SNAPSHOT</version>
+	<name>JGit</name>
     <packaging>jar</packaging>
     <url>http://maven.apache.org</url>
-	<name>JGit</name>
 
     <parent>
         <groupId>com.baeldung</groupId>

README.md

@@ -14,7 +14,7 @@ Java and Spring Tutorials
 ================
 
 This project is **a collection of small and focused tutorials** - each covering a single and well defined area of development in the Java ecosystem. 
-A strong focus of these is, of course, the Spring Framework - Spring, Spring Boot and Spring Securiyt. 
+A strong focus of these is, of course, the Spring Framework - Spring, Spring Boot and Spring Security. 
 In additional to Spring, the following technologies are in focus: `core Java`, `Jackson`, `HttpClient`, `Guava`. 
 
 

Twitter4J/pom.xml

@@ -2,8 +2,8 @@
     xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
     <modelVersion>4.0.0</modelVersion>
     <artifactId>Twitter4J</artifactId>
-    <packaging>jar</packaging>
     <name>Twitter4J</name>
+    <packaging>jar</packaging>
 
     <parent>
         <groupId>com.baeldung</groupId>

akka-http/README.md

@@ -0,0 +1,3 @@
+## Relevant articles:
+
+- [Introduction to Akka HTTP](https://www.baeldung.com/akka-http)

akka-http/pom.xml

@@ -23,7 +23,7 @@
 		<dependency>
 			<groupId>com.typesafe.akka</groupId>
 			<artifactId>akka-stream_2.12</artifactId>
-			<version>2.5.11</version>
+			<version>${akka.stream.version}</version>
 		</dependency>
 		<dependency>
 			<groupId>com.typesafe.akka</groupId>

algorithms-miscellaneous-1/README.md

@@ -12,4 +12,8 @@
 - [Check If a String Contains All The Letters of The Alphabet](https://www.baeldung.com/java-string-contains-all-letters)
 - [Find the Middle Element of a Linked List](http://www.baeldung.com/java-linked-list-middle-element)
 - [Calculate Factorial in Java](https://www.baeldung.com/java-calculate-factorial)
-- [Find Substrings That Are Palindromes in Java](https://www.baeldung.com/java-palindrome-substrings)
+- [Find Substrings That Are Palindromes in Java](https://www.baeldung.com/java-palindrome-substrings)
+- [Find the Longest Substring without Repeating Characters](https://www.baeldung.com/java-longest-substring-without-repeated-characters)
+- [Java Two Pointer Technique](https://www.baeldung.com/java-two-pointer-technique)
+- [Permutations of an Array in Java](https://www.baeldung.com/java-array-permutations)
+- [Implementing Simple State Machines with Java Enums](https://www.baeldung.com/java-enum-simple-state-machine)

algorithms-miscellaneous-1/src/main/java/com/baeldung/algorithms/enumstatemachine/LeaveRequestState.java

@@ -0,0 +1,46 @@
+package com.baeldung.algorithms.enumstatemachine;
+
+public enum LeaveRequestState {
+
+    Submitted {
+        @Override
+        public LeaveRequestState nextState() {
+            System.out.println("Starting the Leave Request and sending to Team Leader for approval.");
+            return Escalated;
+        }
+
+        @Override
+        public String responsiblePerson() {
+            return "Employee";
+        }
+    },
+    Escalated {
+        @Override
+        public LeaveRequestState nextState() {
+            System.out.println("Reviewing the Leave Request and escalating to Department Manager.");
+            return Approved;
+        }
+
+        @Override
+        public String responsiblePerson() {
+            return "Team Leader";
+        }
+    },
+    Approved {
+        @Override
+        public LeaveRequestState nextState() {
+            System.out.println("Approving the Leave Request.");
+            return this;
+        }
+
+        @Override
+        public String responsiblePerson() {
+            return "Department Manager";
+        }
+    };
+
+    public abstract String responsiblePerson();
+
+    public abstract LeaveRequestState nextState();
+
+}

algorithms-miscellaneous-1/src/main/java/com/baeldung/algorithms/permutation/Permutation.java

@@ -0,0 +1,123 @@
+package com.baeldung.algorithms.permutation;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+public  class Permutation {
+
+    public static <T> void printAllRecursive(T[] elements, char delimiter) {
+        printAllRecursive(elements.length, elements, delimiter);
+    }
+
+    public static <T> void printAllRecursive(int n, T[] elements, char delimiter) {
+
+        if(n == 1) {
+            printArray(elements, delimiter);
+        } else {
+            for(int i = 0; i < n-1; i++) {
+                printAllRecursive(n - 1, elements, delimiter);
+                if(n % 2 == 0) {
+                    swap(elements, i, n-1);
+                } else {
+                    swap(elements, 0, n-1);
+                }
+            }
+            printAllRecursive(n - 1, elements, delimiter);
+        }
+    }
+
+    public static <T> void printAllIterative(int n, T[] elements, char delimiter) {
+
+        int[] indexes = new int[n];
+        for (int i = 0; i < n; i++) {
+            indexes[i] = 0;
+        }
+
+        printArray(elements, delimiter);
+
+        int i = 0;
+        while (i < n) {
+            if (indexes[i] < i) {
+                swap(elements, i % 2 == 0 ?  0: indexes[i], i);
+                printArray(elements, delimiter);
+                indexes[i]++;
+                i = 0;
+            }
+            else {
+                indexes[i] = 0;
+                i++;
+            }
+        }
+    }
+
+    public static <T extends Comparable<T>> void printAllOrdered(T[] elements, char delimiter) {
+
+        Arrays.sort(elements);
+        boolean hasNext = true;
+
+        while(hasNext) {
+            printArray(elements, delimiter);
+            int k = 0, l = 0;
+            hasNext = false;
+            for (int i = elements.length - 1; i > 0; i--) {
+                if (elements[i].compareTo(elements[i - 1]) > 0) {
+                    k = i - 1;
+                    hasNext = true;
+                    break;
+                }
+            }
+
+            for (int i = elements.length - 1; i > k; i--) {
+                if (elements[i].compareTo(elements[k]) > 0) {
+                    l = i;
+                    break;
+                }
+            }
+
+            swap(elements, k, l);
+            Collections.reverse(Arrays.asList(elements).subList(k + 1, elements.length));
+        }
+    }
+
+    public static <T> void printRandom(T[] elements, char delimiter) {
+
+        Collections.shuffle(Arrays.asList(elements));
+        printArray(elements, delimiter);
+    }
+
+    private static <T> void swap(T[] elements, int a, int b) {
+
+        T tmp = elements[a];
+        elements[a] = elements[b];
+        elements[b] = tmp;
+    }
+
+    private static <T> void printArray(T[] elements, char delimiter) {
+
+        String delimiterSpace = delimiter + " ";
+        for(int i = 0; i < elements.length; i++) {
+            System.out.print(elements[i] + delimiterSpace);
+        }
+        System.out.print('\n');
+    }
+
+    public static void main(String[] argv) {
+
+        Integer[] elements = {1,2,3,4};
+
+        System.out.println("Rec:");
+        printAllRecursive(elements, ';');
+
+        System.out.println("Iter:");
+        printAllIterative(elements.length, elements, ';');
+
+        System.out.println("Orderes:");
+        printAllOrdered(elements, ';');
+
+        System.out.println("Random:");
+        printRandom(elements, ';');
+
+        System.out.println("Random:");
+        printRandom(elements, ';');
+    }
+}

algorithms-miscellaneous-1/src/main/java/com/baeldung/algorithms/twopointertechnique/LinkedListFindMiddle.java

@@ -0,0 +1,16 @@
+package com.baeldung.algorithms.twopointertechnique;
+
+public class LinkedListFindMiddle {
+
+    public <T> T findMiddle(MyNode<T> head) {
+        MyNode<T> slowPointer = head;
+        MyNode<T> fastPointer = head;
+
+        while (fastPointer.next != null && fastPointer.next.next != null) {
+            fastPointer = fastPointer.next.next;
+            slowPointer = slowPointer.next;
+        }
+        return slowPointer.data;
+    }
+
+}

algorithms-miscellaneous-1/src/main/java/com/baeldung/algorithms/twopointertechnique/MyNode.java

@@ -0,0 +1,20 @@
+package com.baeldung.algorithms.twopointertechnique;
+
+public class MyNode<E> {
+    MyNode<E> next;
+    E data;
+
+    public MyNode(E value) {
+        data = value;
+        next = null;
+    }
+
+    public MyNode(E value, MyNode<E> n) {
+        data = value;
+        next = n;
+    }
+
+    public void setNext(MyNode<E> n) {
+        next = n;
+    }
+}

algorithms-miscellaneous-1/src/main/java/com/baeldung/algorithms/twopointertechnique/RotateArray.java

@@ -0,0 +1,22 @@
+package com.baeldung.algorithms.twopointertechnique;
+
+public class RotateArray {
+
+    public void rotate(int[] input, int step) {
+        step %= input.length;
+        reverse(input, 0, input.length - 1);
+        reverse(input, 0, step - 1);
+        reverse(input, step, input.length - 1);
+    }
+
+    private void reverse(int[] input, int start, int end) {
+        while (start < end) {
+            int temp = input[start];
+            input[start] = input[end];
+            input[end] = temp;
+            start++;
+            end--;
+        }
+    }
+
+}

algorithms-miscellaneous-1/src/main/java/com/baeldung/algorithms/twopointertechnique/TwoSum.java

@@ -0,0 +1,38 @@
+package com.baeldung.algorithms.twopointertechnique;
+
+public class TwoSum {
+
+    public boolean twoSum(int[] input, int targetValue) {
+
+        int pointerOne = 0;
+        int pointerTwo = input.length - 1;
+
+        while (pointerOne < pointerTwo) {
+            int sum = input[pointerOne] + input[pointerTwo];
+
+            if (sum == targetValue) {
+                return true;
+            } else if (sum < targetValue) {
+                pointerOne++;
+            } else {
+                pointerTwo--;
+            }
+        }
+
+        return false;
+    }
+
+    public boolean twoSumSlow(int[] input, int targetValue) {
+
+        for (int i = 0; i < input.length; i++) {
+            for (int j = 1; j < input.length; j++) {
+                if (input[i] + input[j] == targetValue) {
+                    return true;
+                }
+            }
+        }
+
+        return false;
+    }
+
+}

algorithms-miscellaneous-1/src/test/java/com/baeldung/algorithms/enumstatemachine/LeaveRequestStateUnitTest.java

@@ -0,0 +1,37 @@
+package com.baeldung.algorithms.enumstatemachine;
+
+import static org.junit.Assert.assertEquals;
+
+import org.junit.Test;
+
+public class LeaveRequestStateUnitTest {
+
+    @Test
+    public void givenLeaveRequest_whenStateEscalated_thenResponsibleIsTeamLeader() {
+        LeaveRequestState state = LeaveRequestState.Escalated;
+
+        assertEquals(state.responsiblePerson(), "Team Leader");
+    }
+
+
+    @Test
+    public void givenLeaveRequest_whenStateApproved_thenResponsibleIsDepartmentManager() {
+        LeaveRequestState state = LeaveRequestState.Approved;
+
+        assertEquals(state.responsiblePerson(), "Department Manager");
+    }
+
+    @Test
+    public void givenLeaveRequest_whenNextStateIsCalled_thenStateIsChanged() {
+        LeaveRequestState state = LeaveRequestState.Submitted;
+
+        state = state.nextState();
+        assertEquals(state, LeaveRequestState.Escalated);
+
+        state = state.nextState();
+        assertEquals(state, LeaveRequestState.Approved);
+
+        state = state.nextState();
+        assertEquals(state, LeaveRequestState.Approved);
+    }
+}

algorithms-miscellaneous-1/src/test/java/com/baeldung/algorithms/twopointertechnique/LinkedListFindMiddleUnitTest.java

@@ -0,0 +1,37 @@
+package com.baeldung.algorithms.twopointertechnique;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.junit.Test;
+
+public class LinkedListFindMiddleUnitTest {
+
+    LinkedListFindMiddle linkedListFindMiddle = new LinkedListFindMiddle();
+
+    @Test
+    public void givenLinkedListOfMyNodes_whenLinkedListFindMiddle_thenCorrect() {
+
+        MyNode<String> head = createNodesList(8);
+
+        assertThat(linkedListFindMiddle.findMiddle(head)).isEqualTo("4");
+
+        head = createNodesList(9);
+
+        assertThat(linkedListFindMiddle.findMiddle(head)).isEqualTo("5");
+    }
+
+    private static MyNode<String> createNodesList(int n) {
+
+        MyNode<String> head = new MyNode<String>("1");
+        MyNode<String> current = head;
+
+        for (int i = 2; i <= n; i++) {
+            MyNode<String> newNode = new MyNode<String>(String.valueOf(i));
+            current.setNext(newNode);
+            current = newNode;
+        }
+
+        return head;
+    }
+
+}

algorithms-miscellaneous-1/src/test/java/com/baeldung/algorithms/twopointertechnique/RotateArrayUnitTest.java

@@ -0,0 +1,26 @@
+package com.baeldung.algorithms.twopointertechnique;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.junit.Test;
+
+public class RotateArrayUnitTest {
+
+    private RotateArray rotateArray = new RotateArray();
+
+    private int[] inputArray;
+
+    private int step;
+
+    @Test
+    public void givenAnArrayOfIntegers_whenRotateKsteps_thenCorrect() {
+
+        inputArray = new int[] { 1, 2, 3, 4, 5, 6, 7 };
+        step = 4;
+
+        rotateArray.rotate(inputArray, step);
+
+        assertThat(inputArray).containsExactly(new int[] { 4, 5, 6, 7, 1, 2, 3 });
+    }
+
+}

algorithms-miscellaneous-1/src/test/java/com/baeldung/algorithms/twopointertechnique/TwoSumUnitTest.java

@@ -0,0 +1,56 @@
+package com.baeldung.algorithms.twopointertechnique;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import org.junit.Test;
+
+public class TwoSumUnitTest {
+
+    private TwoSum twoSum = new TwoSum();
+
+    private int[] sortedArray;
+
+    private int targetValue;
+
+    @Test
+    public void givenASortedArrayOfIntegers_whenTwoSumSlow_thenPairExists() {
+
+        sortedArray = new int[] { 0, 1, 2, 3, 4, 5, 5, 6, 7, 8, 9, 9 };
+
+        targetValue = 12;
+
+        assertTrue(twoSum.twoSumSlow(sortedArray, targetValue));
+    }
+
+    @Test
+    public void givenASortedArrayOfIntegers_whenTwoSumSlow_thenPairDoesNotExists() {
+
+        sortedArray = new int[] { 0, 1, 2, 3, 4, 5, 5, 6, 7, 8, 9, 9 };
+
+        targetValue = 20;
+
+        assertFalse(twoSum.twoSumSlow(sortedArray, targetValue));
+    }
+
+    @Test
+    public void givenASortedArrayOfIntegers_whenTwoSum_thenPairExists() {
+
+        sortedArray = new int[] { 0, 1, 2, 3, 4, 5, 5, 6, 7, 8, 9, 9 };
+
+        targetValue = 12;
+
+        assertTrue(twoSum.twoSum(sortedArray, targetValue));
+    }
+
+    @Test
+    public void givenASortedArrayOfIntegers_whenTwoSum_thenPairDoesNotExists() {
+
+        sortedArray = new int[] { 0, 1, 2, 3, 4, 5, 5, 6, 7, 8, 9, 9 };
+
+        targetValue = 20;
+
+        assertFalse(twoSum.twoSum(sortedArray, targetValue));
+    }
+
+}

algorithms-miscellaneous-2/pom.xml

@@ -68,7 +68,7 @@
             <plugin>
                 <groupId>org.codehaus.mojo</groupId>
                 <artifactId>cobertura-maven-plugin</artifactId>
-                <version>2.7</version>
+                <version>${cobertura-maven-plugin.version}</version>
                 <configuration>
                     <instrumentation>
                         <ignores>
@@ -91,6 +91,7 @@
         <org.jgrapht.ext.version>1.0.1</org.jgrapht.ext.version>
         <org.assertj.core.version>3.9.0</org.assertj.core.version>
         <commons-codec.version>1.11</commons-codec.version>
+        <cobertura-maven-plugin.version>2.7</cobertura-maven-plugin.version>
     </properties>
 
 </project>

animal-sniffer-mvn-plugin/pom.xml

@@ -3,9 +3,9 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.baeldung</groupId>
     <artifactId>animal-sniffer-mvn-plugin</artifactId>
-    <packaging>jar</packaging>
     <version>1.0-SNAPSHOT</version>
     <name>animal-sniffer-mvn-plugin</name>
+    <packaging>jar</packaging>
     <url>http://maven.apache.org</url>
 
     <parent>

annotations/pom.xml

@@ -3,8 +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>
     <artifactId>annotations</artifactId>
-    <packaging>pom</packaging>
     <name>annotations</name>
+    <packaging>pom</packaging>
     
     <parent>
         <artifactId>parent-modules</artifactId>

apache-avro/pom.xml

@@ -7,14 +7,6 @@
     <version>0.0.1-SNAPSHOT</version>
 	<name>apache-avro</name>
 
-    <properties>
-        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-        <compiler-plugin.version>3.5</compiler-plugin.version>
-        <avro.version>1.8.2</avro.version>
-        <java.version>1.8</java.version>
-        <slf4j.version>1.7.25</slf4j.version>
-    </properties>
-	
     <parent>
         <groupId>com.baeldung</groupId>
         <artifactId>parent-modules</artifactId>
@@ -85,4 +77,13 @@
             </plugin>
         </plugins>
     </build>
+
+    <properties>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+        <compiler-plugin.version>3.5</compiler-plugin.version>
+        <avro.version>1.8.2</avro.version>
+        <java.version>1.8</java.version>
+        <slf4j.version>1.7.25</slf4j.version>
+    </properties>
+	
 </project>

apache-curator/pom.xml

@@ -3,8 +3,8 @@
     <modelVersion>4.0.0</modelVersion>
     <artifactId>apache-curator</artifactId>
     <version>0.0.1-SNAPSHOT</version>
-    <packaging>jar</packaging>
 	<name>apache-curator</name>
+    <packaging>jar</packaging>
 
     <parent>
         <groupId>com.baeldung</groupId>

apache-geode/pom.xml

@@ -12,22 +12,6 @@
         <artifactId>parent-modules</artifactId>
         <version>1.0.0-SNAPSHOT</version>
     </parent>
-    
-    <properties>
-        <geode.core>1.6.0</geode.core>
-    </properties>
-    <build>
-        <plugins>
-            <plugin>
-                <groupId>org.apache.maven.plugins</groupId>
-                <artifactId>maven-compiler-plugin</artifactId>
-                <configuration>
-                    <source>1.8</source>
-                    <target>1.8</target>
-                </configuration>
-            </plugin>
-        </plugins>
-    </build>
 
     <dependencies>
         <dependency>
@@ -38,8 +22,26 @@
         <dependency>
             <groupId>junit</groupId>
             <artifactId>junit</artifactId>
-            <version>RELEASE</version>
+            <version>${junit.version}</version>
         </dependency>
     </dependencies>
     
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <source>${maven.compiler.source}</source>
+                    <target>${maven.compiler.target}</target>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+	
+    <properties>
+        <geode.core>1.6.0</geode.core>
+        <maven.compiler.source>1.8</maven.compiler.source>
+        <maven.compiler.target>1.8</maven.compiler.target>
+    </properties>
 </project>

apache-meecrowave/pom.xml

@@ -7,51 +7,58 @@
     <name>apache-meecrowave</name>
     <description>A sample REST API application with Meecrowave</description>
 
-    <properties>
-        <maven.compiler.source>1.8</maven.compiler.source>
-        <maven.compiler.target>1.8</maven.compiler.target>
-    </properties>
     <dependencies>
         <!-- https://mvnrepository.com/artifact/org.apache.meecrowave/meecrowave-core -->
         <dependency>
             <groupId>org.apache.meecrowave</groupId>
             <artifactId>meecrowave-core</artifactId>
-            <version>1.2.1</version>
+            <version>${meecrowave-core.version}</version>
         </dependency>
         <!-- https://mvnrepository.com/artifact/org.apache.meecrowave/meecrowave-jpa -->
         <dependency>
             <groupId>org.apache.meecrowave</groupId>
             <artifactId>meecrowave-jpa</artifactId>
-            <version>1.2.1</version>
+            <version>${meecrowave-jpa.version}</version>
         </dependency>
 
         <dependency>
             <groupId>com.squareup.okhttp3</groupId>
             <artifactId>okhttp</artifactId>
-            <version>3.10.0</version>
+            <version>${okhttp.version}</version>
         </dependency>
         <dependency>
             <groupId>org.apache.meecrowave</groupId>
             <artifactId>meecrowave-junit</artifactId>
-            <version>1.2.0</version>
+            <version>${meecrowave-junit.version}</version>
             <scope>test</scope>
         </dependency>
         <!-- https://mvnrepository.com/artifact/junit/junit -->
         <dependency>
             <groupId>junit</groupId>
             <artifactId>junit</artifactId>
-            <version>4.10</version>
+            <version>${junit.version}</version>
             <scope>test</scope>
         </dependency>
-
     </dependencies>
+	
     <build>
         <plugins>
             <plugin>
                 <groupId>org.apache.meecrowave</groupId>
                 <artifactId>meecrowave-maven-plugin</artifactId>
-                <version>1.2.1</version>
+                <version>${meecrowave-maven-plugin.version}</version>
             </plugin>
         </plugins>
     </build>
+	
+    <properties>
+        <maven.compiler.source>1.8</maven.compiler.source>
+        <maven.compiler.target>1.8</maven.compiler.target>
+		<junit.version>4.10</junit.version>
+		<meecrowave-junit.version>1.2.0</meecrowave-junit.version>
+		<okhttp.version>3.10.0</okhttp.version>
+		<meecrowave-jpa.version>1.2.1</meecrowave-jpa.version>
+		<meecrowave-core.version>1.2.1</meecrowave-core.version>
+		<meecrowave-maven-plugin.version>1.2.1</meecrowave-maven-plugin.version>
+    </properties>
 </project>

apache-pulsar/pom.xml

@@ -11,12 +11,14 @@
         <dependency>
             <groupId>org.apache.pulsar</groupId>
             <artifactId>pulsar-client</artifactId>
-            <version>2.1.1-incubating</version>
+            <version>${pulsar-client.version}</version>
             <scope>compile</scope>
         </dependency>
     </dependencies>
+	
     <properties>
         <maven.compiler.target>1.8</maven.compiler.target>
         <maven.compiler.source>1.8</maven.compiler.source>
+		<pulsar-client.version>2.1.1-incubating</pulsar-client.version>
     </properties>
 </project>

apache-solrj/pom.xml

@@ -4,8 +4,8 @@
     <groupId>com.baeldung</groupId>
     <artifactId>apache-solrj</artifactId>
     <version>0.0.1-SNAPSHOT</version>
-    <packaging>jar</packaging>
     <name>apache-solrj</name>
+    <packaging>jar</packaging>
 
     <parent>
         <groupId>com.baeldung</groupId>

apache-spark/README.md

@@ -1,3 +1,4 @@
 ### Relevant articles
 
 - [Introduction to Apache Spark](http://www.baeldung.com/apache-spark)
+- [Building a Data Pipeline with Kafka, Spark Streaming and Cassandra](https://www.baeldung.com/kafka-spark-data-pipeline)

apache-spark/pom.xml

@@ -4,8 +4,8 @@
     <groupId>com.baeldung</groupId>
     <artifactId>apache-spark</artifactId>
     <version>1.0-SNAPSHOT</version>
-    <packaging>jar</packaging>
     <name>apache-spark</name>
+    <packaging>jar</packaging>
     <url>http://maven.apache.org</url>
 
     <parent>
@@ -15,16 +15,79 @@
     </parent>
 
     <dependencies>
-        <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-core_2.10 -->
         <dependency>
             <groupId>org.apache.spark</groupId>
-            <artifactId>spark-core_2.10</artifactId>
+            <artifactId>spark-core_2.11</artifactId>
             <version>${org.apache.spark.spark-core.version}</version>
+            <scope>provided</scope>
         </dependency>
+		<dependency>
+    		<groupId>org.apache.spark</groupId>
+    		<artifactId>spark-sql_2.11</artifactId>
+    		<version>${org.apache.spark.spark-sql.version}</version>
+    		<scope>provided</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.spark</groupId>
+			<artifactId>spark-streaming_2.11</artifactId>
+			<version>${org.apache.spark.spark-streaming.version}</version>
+			<scope>provided</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.spark</groupId>
+			<artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
+			<version>${org.apache.spark.spark-streaming-kafka.version}</version>
+		</dependency>
+		<dependency>
+			<groupId>com.datastax.spark</groupId>
+			<artifactId>spark-cassandra-connector_2.11</artifactId>
+			<version>${com.datastax.spark.spark-cassandra-connector.version}</version>
+		</dependency>
+		<dependency>
+			<groupId>com.datastax.spark</groupId>
+			<artifactId>spark-cassandra-connector-java_2.11</artifactId>
+			<version>${com.datastax.spark.spark-cassandra-connector-java.version}</version>
+		</dependency>
     </dependencies>
-
+	<build>
+    	<plugins>
+    	  	<plugin>
+    			<groupId>org.apache.maven.plugins</groupId>
+    			<artifactId>maven-compiler-plugin</artifactId>
+    			<version>${maven-compiler-plugin.version}</version>
+    			<configuration>
+      				<source>${maven.compiler.source}</source>
+      				<target>${maven.compiler.target}</target>
+    			</configuration>
+  			</plugin>
+      		<plugin>
+        		<artifactId>maven-assembly-plugin</artifactId>
+        		<executions>
+          			<execution>
+            			<phase>package</phase>
+            			<goals>
+              				<goal>single</goal>
+            			</goals>
+          			</execution>
+        		</executions>
+        		<configuration>
+          			<descriptorRefs>
+            			<descriptorRef>jar-with-dependencies</descriptorRef>
+          			</descriptorRefs>
+        		</configuration>
+      		</plugin>
+   		</plugins>
+  	</build>
     <properties>
-        <org.apache.spark.spark-core.version>2.2.0</org.apache.spark.spark-core.version>
+        <org.apache.spark.spark-core.version>2.3.0</org.apache.spark.spark-core.version>
+        <org.apache.spark.spark-sql.version>2.3.0</org.apache.spark.spark-sql.version>
+        <org.apache.spark.spark-streaming.version>2.3.0</org.apache.spark.spark-streaming.version>
+        <org.apache.spark.spark-streaming-kafka.version>2.3.0</org.apache.spark.spark-streaming-kafka.version>
+        <com.datastax.spark.spark-cassandra-connector.version>2.3.0</com.datastax.spark.spark-cassandra-connector.version>
+        <com.datastax.spark.spark-cassandra-connector-java.version>1.5.2</com.datastax.spark.spark-cassandra-connector-java.version>
+        <maven.compiler.source>1.8</maven.compiler.source>
+        <maven.compiler.target>1.8</maven.compiler.target>
+		<maven-compiler-plugin.version>3.2</maven-compiler-plugin.version>
     </properties>
 
 </project>

apache-spark/src/main/java/com/baeldung/data/pipeline/Word.java

@@ -0,0 +1,25 @@
+package com.baeldung.data.pipeline;
+
+import java.io.Serializable;
+
+public class Word implements Serializable {
+    private static final long serialVersionUID = 1L;
+    private String word;
+    private int count;
+    Word(String word, int count) {
+        this.word = word;
+        this.count = count;
+    }
+    public String getWord() {
+        return word;
+    }
+    public void setWord(String word) {
+        this.word = word;
+    }
+    public int getCount() {
+        return count;
+    }
+    public void setCount(int count) {
+        this.count = count;
+    }
+}

apache-spark/src/main/java/com/baeldung/data/pipeline/WordCountingApp.java

@@ -0,0 +1,80 @@
+package com.baeldung.data.pipeline;
+
+import static com.datastax.spark.connector.japi.CassandraJavaUtil.javaFunctions;
+import static com.datastax.spark.connector.japi.CassandraJavaUtil.mapToRow;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.log4j.Level;
+import org.apache.log4j.Logger;
+import org.apache.spark.SparkConf;
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.streaming.Durations;
+import org.apache.spark.streaming.api.java.JavaDStream;
+import org.apache.spark.streaming.api.java.JavaInputDStream;
+import org.apache.spark.streaming.api.java.JavaPairDStream;
+import org.apache.spark.streaming.api.java.JavaStreamingContext;
+import org.apache.spark.streaming.kafka010.ConsumerStrategies;
+import org.apache.spark.streaming.kafka010.KafkaUtils;
+import org.apache.spark.streaming.kafka010.LocationStrategies;
+
+import scala.Tuple2;
+
+public class WordCountingApp {
+
+    public static void main(String[] args) throws InterruptedException {
+        Logger.getLogger("org")
+            .setLevel(Level.OFF);
+        Logger.getLogger("akka")
+            .setLevel(Level.OFF);
+
+        Map<String, Object> kafkaParams = new HashMap<>();
+        kafkaParams.put("bootstrap.servers", "localhost:9092");
+        kafkaParams.put("key.deserializer", StringDeserializer.class);
+        kafkaParams.put("value.deserializer", StringDeserializer.class);
+        kafkaParams.put("group.id", "use_a_separate_group_id_for_each_stream");
+        kafkaParams.put("auto.offset.reset", "latest");
+        kafkaParams.put("enable.auto.commit", false);
+
+        Collection<String> topics = Arrays.asList("messages");
+
+        SparkConf sparkConf = new SparkConf();
+        sparkConf.setMaster("local[2]");
+        sparkConf.setAppName("WordCountingApp");
+        sparkConf.set("spark.cassandra.connection.host", "127.0.0.1");
+
+        JavaStreamingContext streamingContext = new JavaStreamingContext(sparkConf, Durations.seconds(1));
+
+        JavaInputDStream<ConsumerRecord<String, String>> messages = KafkaUtils.createDirectStream(streamingContext, LocationStrategies.PreferConsistent(), ConsumerStrategies.<String, String> Subscribe(topics, kafkaParams));
+
+        JavaPairDStream<String, String> results = messages.mapToPair(record -> new Tuple2<>(record.key(), record.value()));
+
+        JavaDStream<String> lines = results.map(tuple2 -> tuple2._2());
+
+        JavaDStream<String> words = lines.flatMap(x -> Arrays.asList(x.split("\\s+"))
+            .iterator());
+
+        JavaPairDStream<String, Integer> wordCounts = words.mapToPair(s -> new Tuple2<>(s, 1))
+            .reduceByKey((i1, i2) -> i1 + i2);
+
+        wordCounts.foreachRDD(javaRdd -> {
+            Map<String, Integer> wordCountMap = javaRdd.collectAsMap();
+            for (String key : wordCountMap.keySet()) {
+                List<Word> wordList = Arrays.asList(new Word(key, wordCountMap.get(key)));
+                JavaRDD<Word> rdd = streamingContext.sparkContext()
+                    .parallelize(wordList);
+                javaFunctions(rdd).writerBuilder("vocabulary", "words", mapToRow(Word.class))
+                    .saveToCassandra();
+            }
+        });
+
+        streamingContext.start();
+        streamingContext.awaitTermination();
+    }
+}