Merge branch 'master' of https://github.com/JonCook/tutorials

libraries-data/src/main/java/com/baeldung/crunch/StopWordFilter.java

@@ -0,0 +1,25 @@
+package com.baeldung.crunch;
+
+import java.util.Set;
+
+import org.apache.crunch.FilterFn;
+
+import com.google.common.collect.ImmutableSet;
+
+/**
+ * A filter that removes known stop words.
+ */
+public class StopWordFilter extends FilterFn<String> {
+
+    // English stop words, borrowed from Lucene.
+    private static final Set<String> STOP_WORDS = ImmutableSet
+        .copyOf(new String[] { "a", "and", "are", "as", "at", "be", "but", "by",
+                "for", "if", "in", "into", "is", "it", "no", "not", "of", "on",
+                "or", "s", "such", "t", "that", "the", "their", "then", "there",
+                "these", "they", "this", "to", "was", "will", "with" });
+
+    @Override
+    public boolean accept(String word) {
+        return !STOP_WORDS.contains(word);
+    }
+}

libraries-data/src/main/java/com/baeldung/crunch/ToUpperCaseFn.java

@@ -0,0 +1,11 @@
+package com.baeldung.crunch;
+
+import org.apache.crunch.MapFn;
+
+public class ToUpperCaseFn extends MapFn<String, String> {
+
+    @Override
+    public String map(String input) {
+        return input != null ? input.toUpperCase() : input;
+    }
+}

libraries-data/src/main/java/com/baeldung/crunch/ToUpperCaseWithCounterFn.java

@@ -0,0 +1,20 @@
+package com.baeldung.crunch;
+
+import org.apache.crunch.MapFn;
+
+@SuppressWarnings("serial")
+public class ToUpperCaseWithCounterFn extends MapFn<String, String> {
+
+    @Override
+    public String map(String input) {
+        if (input == null) {
+            return input;
+        } else {
+            String output = input.toUpperCase();
+            if (!input.equals(output)) {
+                increment("UpperCase", "modified");
+            }
+            return output;
+        }
+    }
+}

libraries-data/src/main/java/com/baeldung/crunch/Tokenizer.java

@@ -0,0 +1,23 @@
+package com.baeldung.crunch;
+
+import org.apache.crunch.DoFn;
+import org.apache.crunch.Emitter;
+
+import com.google.common.base.Splitter;
+
+/**
+ * Splits a line of text, filtering known stop words.
+ */
+public class Tokenizer extends DoFn<String, String> {
+    private static final Splitter SPLITTER = Splitter
+        .onPattern("\\s+")
+        .omitEmptyStrings();
+
+    @Override
+    public void process(String line,
+        Emitter<String> emitter) {
+        for (String word : SPLITTER.split(line)) {
+            emitter.emit(word);
+        }
+    }
+}

libraries-data/src/main/java/com/baeldung/crunch/WordCount.java

@@ -0,0 +1,62 @@
+package com.baeldung.crunch;
+
+import org.apache.crunch.PCollection;
+import org.apache.crunch.PTable;
+import org.apache.crunch.Pipeline;
+import org.apache.crunch.PipelineResult;
+import org.apache.crunch.impl.mr.MRPipeline;
+import org.apache.crunch.types.writable.Writables;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.conf.Configured;
+import org.apache.hadoop.util.GenericOptionsParser;
+import org.apache.hadoop.util.Tool;
+import org.apache.hadoop.util.ToolRunner;
+
+/**
+ * A word count example for Apache Crunch, based on Crunch's example projects.
+ */
+public class WordCount extends Configured implements Tool {
+
+    public static void main(String[] args) throws Exception {
+        ToolRunner.run(new Configuration(), new WordCount(), args);
+    }
+
+    public int run(String[] args) throws Exception {
+
+        if (args.length != 2) {
+            System.err.println("Usage: hadoop jar crunch-1.0.0-SNAPSHOT-job.jar" + " [generic options] input output");
+            System.err.println();
+            GenericOptionsParser.printGenericCommandUsage(System.err);
+            return 1;
+        }
+
+        String inputPath = args[0];
+        String outputPath = args[1];
+
+        // Create an object to coordinate pipeline creation and execution.
+        Pipeline pipeline = new MRPipeline(WordCount.class, getConf());
+
+        // Reference a given text file as a collection of Strings.
+        PCollection<String> lines = pipeline.readTextFile(inputPath);
+
+        // Define a function that splits each line in a PCollection of Strings into
+        // a PCollection made up of the individual words in the file.
+        // The second argument sets the serialization format.
+        PCollection<String> words = lines.parallelDo(new Tokenizer(), Writables.strings());
+
+        // Take the collection of words and remove known stop words.
+        PCollection<String> noStopWords = words.filter(new StopWordFilter());
+
+        // The count method applies a series of Crunch primitives and returns
+        // a map of the unique words in the input PCollection to their counts.
+        PTable<String, Long> counts = noStopWords.count();
+
+        // Instruct the pipeline to write the resulting counts to a text file.
+        pipeline.writeTextFile(counts, outputPath);
+
+        // Execute the pipeline as a MapReduce.
+        PipelineResult result = pipeline.done();
+
+        return result.succeeded() ? 0 : 1;
+    }
+}

libraries-data/src/main/java/com/baeldung/ignite/cache/CustomLifecycleBean.java

@@ -0,0 +1,14 @@
+package com.baeldung.ignite.cache;
+
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.lifecycle.LifecycleBean;
+import org.apache.ignite.lifecycle.LifecycleEventType;
+
+public class CustomLifecycleBean implements LifecycleBean {
+    @Override
+    public void onLifecycleEvent(LifecycleEventType lifecycleEventType) throws IgniteException {
+        if (lifecycleEventType == LifecycleEventType.AFTER_NODE_START) {
+            //do something right after the Ignite node starts
+        }
+    }
+}

libraries-data/src/main/java/com/baeldung/ignite/cache/IgniteCacheExample.java

@@ -0,0 +1,58 @@
+package com.baeldung.ignite.cache;
+
+import com.baeldung.ignite.model.Employee;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.Ignition;
+import org.apache.ignite.cache.query.QueryCursor;
+import org.apache.ignite.cache.query.SqlFieldsQuery;
+import org.apache.ignite.configuration.IgniteConfiguration;
+
+import java.util.List;
+
+public class IgniteCacheExample {
+
+    public static void main(String[] args) {
+
+        Ignite ignite = Ignition.ignite();
+
+        IgniteCache<Integer, String> cache = ignite.cache("baeldungCache");
+
+        cache.put(1, "baeldung cache value");
+
+        String message = cache.get(1);
+    }
+
+    private static void getObjectFromCache(Ignite ignite) {
+
+        IgniteCache<Integer, Employee> cache = ignite.getOrCreateCache("baeldungCache");
+
+        cache.put(1, new Employee(1, "John", true));
+        cache.put(2, new Employee(2, "Anna", false));
+        cache.put(3, new Employee(3, "George", true));
+
+        Employee employee = cache.get(1);
+    }
+
+    private static void getFromCacheWithSQl(Ignite ignite) {
+
+        IgniteCache<Integer, Employee> cache = ignite.cache("baeldungCache");
+
+        SqlFieldsQuery sql = new SqlFieldsQuery(
+                "select name from Employee where isEmployed = 'true'");
+
+        QueryCursor<List<?>> cursor = cache.query(sql);
+
+        for (List<?> row : cursor) {
+            System.out.println(row.get(0));
+        }
+    }
+
+    private static void customInitialization() {
+
+        IgniteConfiguration configuration = new IgniteConfiguration();
+        configuration.setLifecycleBeans(new CustomLifecycleBean());
+        Ignite ignite = Ignition.start(configuration);
+    }
+
+}

libraries-data/src/main/java/com/baeldung/ignite/jdbc/IgniteJDBC.java

@@ -0,0 +1,58 @@
+package com.baeldung.ignite.jdbc;
+
+import java.sql.*;
+
+/**
+ * Created by Gebruiker on 3/14/2018.
+ */
+public class IgniteJDBC {
+
+    public static void main(String[] args) throws ClassNotFoundException, SQLException {
+
+        Class.forName("org.apache.ignite.IgniteJdbcThinDriver");
+
+        Connection conn = DriverManager.getConnection("jdbc:ignite:thin://127.0.0.1/");
+
+        createDatabaseTables(conn);
+
+        insertData(conn);
+
+        getData(conn);
+    }
+
+    private static void createDatabaseTables(Connection conn) throws SQLException {
+
+        Statement sql = conn.createStatement();
+        sql.executeUpdate("CREATE TABLE Employee (" +
+                " id INTEGER PRIMARY KEY, name VARCHAR, isEmployed timyint(1)) " +
+                " WITH \"template=replicated\"");
+
+        sql.executeUpdate("CREATE INDEX idx_employee_name ON Employee (name)");
+    }
+
+    private static void insertData(Connection conn) throws SQLException {
+
+        PreparedStatement sql =
+                conn.prepareStatement("INSERT INTO Employee (id, name, isEmployed) VALUES (?, ?, ?)");
+        sql.setLong(1, 1);
+        sql.setString(2, "James");
+        sql.setBoolean(3, true);
+        sql.executeUpdate();
+
+        sql.setLong(1, 2);
+        sql.setString(2, "Monica");
+        sql.setBoolean(3, false);
+        sql.executeUpdate();
+    }
+
+    private static void getData(Connection conn) throws SQLException {
+
+        Statement sql = conn.createStatement();
+        ResultSet rs = sql.executeQuery("SELECT e.name, e.isEmployed " +
+                " FROM Employee e " +
+                " WHERE e.isEmployed = TRUE ");
+
+        while (rs.next())
+            System.out.println(rs.getString(1) + ", " + rs.getString(2));
+    }
+}

libraries-data/src/main/java/com/baeldung/ignite/model/Employee.java

@@ -0,0 +1,48 @@
+package com.baeldung.ignite.model;
+
+
+public class Employee {
+
+    private Integer id;
+    private String name;
+    private boolean isEmployed;
+
+    public Employee(Integer id, String name, boolean isEmployed) {
+        this.id = id;
+        this.name = name;
+        this.isEmployed = isEmployed;
+    }
+
+    public Integer getId() {
+        return id;
+    }
+
+    public void setId(Integer id) {
+        this.id = id;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public void setName(String name) {
+        this.name = name;
+    }
+
+    public boolean isEmployed() {
+        return isEmployed;
+    }
+
+    public void setEmployed(boolean employed) {
+        isEmployed = employed;
+    }
+
+    @Override
+    public String toString() {
+        return "Employee{" +
+                "id=" + id +
+                ", name='" + name + '\'' +
+                ", isEmployed=" + isEmployed +
+                '}';
+    }
+}

libraries-data/src/main/java/com/baeldung/ignite/spring/IgniteApp.java

@@ -0,0 +1,53 @@
+package com.baeldung.ignite.spring;
+
+import com.baeldung.ignite.spring.config.SpringDataConfig;
+import com.baeldung.ignite.spring.dto.EmployeeDTO;
+import com.baeldung.ignite.spring.repository.EmployeeRepository;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.Ignition;
+import org.apache.ignite.cache.query.QueryCursor;
+import org.apache.ignite.cache.query.SqlFieldsQuery;
+import org.springframework.context.annotation.AnnotationConfigApplicationContext;
+
+import java.util.List;
+
+/**
+ * Created by Gebruiker on 4/12/2018.
+ */
+public class IgniteApp {
+
+    public static void main (String[] args) {
+        AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
+        context.register(SpringDataConfig.class);
+        context.refresh();
+
+        EmployeeRepository repository = context.getBean(EmployeeRepository.class);
+
+        EmployeeDTO employeeDTO = new EmployeeDTO();
+        employeeDTO.setId(1);
+        employeeDTO.setName("John");
+        employeeDTO.setEmployed(true);
+
+        repository.save(employeeDTO.getId(), employeeDTO);
+
+        EmployeeDTO employee = repository.getEmployeeDTOById(employeeDTO.getId());
+        System.out.println(employee);
+    }
+
+    private void getUsingTheCache(Integer employeeId) {
+
+        Ignite ignite = Ignition.ignite();
+
+        IgniteCache<Integer, EmployeeDTO> cache = ignite.cache("baeldungCache");
+
+        EmployeeDTO employeeDTO = cache.get(employeeId);
+
+        System.out.println(employeeDTO);
+
+        SqlFieldsQuery sql = new SqlFieldsQuery(
+                "select * from EmployeeDTO where isEmployed = 'true'");
+
+        QueryCursor<List<?>> cursor = cache.query(sql);
+    }
+}

libraries-data/src/main/java/com/baeldung/ignite/spring/config/SpringDataConfig.java

@@ -0,0 +1,29 @@
+package com.baeldung.ignite.spring.config;
+
+import com.baeldung.ignite.spring.dto.EmployeeDTO;
+import com.baeldung.ignite.spring.repository.EmployeeRepository;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.Ignition;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.springdata.repository.config.EnableIgniteRepositories;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.ComponentScan;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+@EnableIgniteRepositories(basePackageClasses = EmployeeRepository.class)
+@ComponentScan(basePackages = "com.baeldung.ignite.spring.repository")
+public class SpringDataConfig {
+
+    @Bean
+    public Ignite igniteInstance() {
+        IgniteConfiguration config = new IgniteConfiguration();
+        
+        CacheConfiguration cache = new CacheConfiguration("baeldungCache");
+        
+        cache.setIndexedTypes(Integer.class, EmployeeDTO.class);
+        config.setCacheConfiguration(cache);
+        return Ignition.start(config);
+    }
+}

libraries-data/src/main/java/com/baeldung/ignite/spring/dto/EmployeeDTO.java

@@ -0,0 +1,48 @@
+package com.baeldung.ignite.spring.dto;
+
+import org.apache.ignite.cache.query.annotations.QuerySqlField;
+
+import java.io.Serializable;
+
+public class EmployeeDTO implements Serializable {
+
+    @QuerySqlField(index = true)
+    private Integer id;
+    @QuerySqlField(index = true)
+    private String name;
+    @QuerySqlField(index = true)
+    private boolean isEmployed;
+
+    public Integer getId() {
+        return id;
+    }
+
+    public void setId(Integer id) {
+        this.id = id;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public void setName(String name) {
+        this.name = name;
+    }
+
+    public boolean isEmployed() {
+        return isEmployed;
+    }
+
+    public void setEmployed(boolean employed) {
+        isEmployed = employed;
+    }
+
+    @Override
+    public String toString() {
+        return "EmployeeDTO{" +
+                "id=" + id +
+                ", name='" + name + '\'' +
+                ", isEmployed=" + isEmployed +
+                '}';
+    }
+}

libraries-data/src/main/java/com/baeldung/ignite/spring/repository/EmployeeRepository.java

@@ -0,0 +1,13 @@
+package com.baeldung.ignite.spring.repository;
+
+import com.baeldung.ignite.spring.dto.EmployeeDTO;
+import org.apache.ignite.springdata.repository.IgniteRepository;
+import org.apache.ignite.springdata.repository.config.RepositoryConfig;
+import org.springframework.stereotype.Repository;
+
+@Repository
+@RepositoryConfig(cacheName = "baeldungCache")
+public interface EmployeeRepository extends IgniteRepository<EmployeeDTO, Integer> {
+
+    EmployeeDTO getEmployeeDTOById(Integer id);
+}

libraries-data/src/main/java/com/baeldung/ignite/stream/CacheConfig.java

@@ -0,0 +1,24 @@
+package com.baeldung.ignite.stream;
+
+import com.baeldung.ignite.model.Employee;
+import org.apache.ignite.configuration.CacheConfiguration;
+
+import javax.cache.configuration.FactoryBuilder;
+import javax.cache.expiry.CreatedExpiryPolicy;
+import javax.cache.expiry.Duration;
+import java.util.concurrent.TimeUnit;
+
+
+public class CacheConfig {
+
+    public static CacheConfiguration<Integer, Employee> employeeCache() {
+
+        CacheConfiguration<Integer, Employee> config = new CacheConfiguration<>("baeldungEmployees");
+
+        config.setIndexedTypes(Integer.class, Employee.class);
+        config.setExpiryPolicyFactory(FactoryBuilder.factoryOf(
+                new CreatedExpiryPolicy(new Duration(TimeUnit.SECONDS, 5))));
+
+        return config;
+    }
+}

libraries-data/src/main/java/com/baeldung/ignite/stream/IgniteStream.java

@@ -0,0 +1,46 @@
+package com.baeldung.ignite.stream;
+
+import com.baeldung.ignite.model.Employee;
+import com.google.gson.Gson;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteDataStreamer;
+import org.apache.ignite.Ignition;
+import org.apache.ignite.stream.StreamTransformer;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+
+public class IgniteStream {
+
+    private static final Gson GSON = new Gson();
+
+    public static void main(String[] args) throws Exception {
+
+        Ignition.setClientMode(true);
+        Ignite ignite = Ignition.start();
+
+        IgniteCache<Integer, Employee> cache = ignite.getOrCreateCache(CacheConfig.employeeCache());
+        IgniteDataStreamer<Integer, Employee> streamer = ignite.dataStreamer(cache.getName());
+        streamer.allowOverwrite(true);
+
+        streamer.receiver(StreamTransformer.from((e, arg) -> {
+
+            Employee employee = e.getValue();
+            employee.setEmployed(true);
+            e.setValue(employee);
+
+            return employee;
+        }));
+
+        Path path = Paths.get(IgniteStream.class.getResource("employees.txt").toURI());
+
+        Files.lines(path)
+          .forEach(line -> {
+              Employee employee = GSON.fromJson(line, Employee.class);
+              streamer.addData(employee.getId(), employee);
+          });
+    }    
+    
+}

libraries-data/src/main/java/com/baeldung/jcache/SimpleCacheEntryListener.java

@@ -0,0 +1,34 @@
+package com.baeldung.jcache;
+
+import java.io.Serializable;
+
+import javax.cache.event.CacheEntryCreatedListener;
+import javax.cache.event.CacheEntryEvent;
+import javax.cache.event.CacheEntryListenerException;
+import javax.cache.event.CacheEntryUpdatedListener;
+
+public class SimpleCacheEntryListener implements CacheEntryCreatedListener<String, String>, CacheEntryUpdatedListener<String, String>, Serializable {
+
+    /**
+     * 
+     */
+    private static final long serialVersionUID = -712657810462878763L;
+    private boolean updated;
+    private boolean created;
+
+    public boolean getUpdated() {
+        return this.updated;
+    }
+
+    public boolean getCreated() {
+        return this.created;
+    }
+
+    public void onUpdated(Iterable<CacheEntryEvent<? extends String, ? extends String>> events) throws CacheEntryListenerException {
+        this.updated = true;
+    }
+
+    public void onCreated(Iterable<CacheEntryEvent<? extends String, ? extends String>> events) throws CacheEntryListenerException {
+        this.created = true;
+    }
+}

libraries-data/src/main/java/com/baeldung/jcache/SimpleCacheLoader.java

@@ -0,0 +1,24 @@
+package com.baeldung.jcache;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import javax.cache.integration.CacheLoader;
+import javax.cache.integration.CacheLoaderException;
+
+public class SimpleCacheLoader implements CacheLoader<Integer, String> {
+
+    @Override
+    public String load(Integer key) throws CacheLoaderException {
+        return "fromCache" + key;
+    }
+
+    @Override
+    public Map<Integer, String> loadAll(Iterable<? extends Integer> keys) throws CacheLoaderException {
+        Map<Integer, String> data = new HashMap<>();
+        for (int key : keys) {
+            data.put(key, load(key));
+        }
+        return data;
+    }
+}

libraries-data/src/main/java/com/baeldung/jcache/SimpleEntryProcessor.java

@@ -0,0 +1,25 @@
+package com.baeldung.jcache;
+
+import java.io.Serializable;
+
+import javax.cache.processor.EntryProcessor;
+import javax.cache.processor.EntryProcessorException;
+import javax.cache.processor.MutableEntry;
+
+public class SimpleEntryProcessor implements EntryProcessor<String, String, String>, Serializable {
+
+    /**
+     * 
+     */
+    private static final long serialVersionUID = -5616476363722945132L;
+
+    public String process(MutableEntry<String, String> entry, Object... args) throws EntryProcessorException {
+
+        if (entry.exists()) {
+            String current = entry.getValue();
+            entry.setValue(current + " - modified");
+            return current;
+        }
+        return null;
+    }
+}

libraries-data/src/main/java/com/baeldung/storm/TopologyRunner.java

@@ -0,0 +1,34 @@
+package com.baeldung.storm;
+
+import com.baeldung.storm.bolt.AggregatingBolt;
+import com.baeldung.storm.bolt.FileWritingBolt;
+import com.baeldung.storm.bolt.FilteringBolt;
+import com.baeldung.storm.bolt.PrintingBolt;
+import com.baeldung.storm.spout.RandomNumberSpout;
+import org.apache.storm.Config;
+import org.apache.storm.LocalCluster;
+import org.apache.storm.topology.TopologyBuilder;
+import org.apache.storm.topology.base.BaseWindowedBolt;
+
+public class TopologyRunner {
+    public static void main(String[] args) {
+       runTopology();
+    }
+
+    public static void runTopology() {
+        String filePath = "./src/main/resources/operations.txt";
+        TopologyBuilder builder = new TopologyBuilder();
+        builder.setSpout("randomNumberSpout", new RandomNumberSpout());
+        builder.setBolt("filteringBolt", new FilteringBolt()).shuffleGrouping("randomNumberSpout");
+        builder.setBolt("aggregatingBolt", new AggregatingBolt()
+          .withTimestampField("timestamp")
+          .withLag(BaseWindowedBolt.Duration.seconds(1))
+          .withWindow(BaseWindowedBolt.Duration.seconds(5))).shuffleGrouping("filteringBolt");
+        builder.setBolt("fileBolt", new FileWritingBolt(filePath)).shuffleGrouping("aggregatingBolt");
+
+        Config config = new Config();
+        config.setDebug(false);
+        LocalCluster cluster = new LocalCluster();
+        cluster.submitTopology("Test", config, builder.createTopology());
+    }
+}

libraries-data/src/main/java/com/baeldung/storm/bolt/AggregatingBolt.java

@@ -0,0 +1,39 @@
+package com.baeldung.storm.bolt;
+
+import org.apache.storm.task.OutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.topology.base.BaseWindowedBolt;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Tuple;
+import org.apache.storm.tuple.Values;
+import org.apache.storm.windowing.TupleWindow;
+
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+
+public class AggregatingBolt extends BaseWindowedBolt {
+    private OutputCollector outputCollector;
+    @Override
+    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
+        this.outputCollector = collector;
+    }
+
+    @Override
+    public void declareOutputFields(OutputFieldsDeclarer declarer) {
+        declarer.declare(new Fields("sumOfOperations", "beginningTimestamp", "endTimestamp"));
+    }
+
+    @Override
+    public void execute(TupleWindow tupleWindow) {
+        List<Tuple> tuples = tupleWindow.get();
+        tuples.sort(Comparator.comparing(a -> a.getLongByField("timestamp")));
+        //This is safe since the window is calculated basing on Tuple's timestamp, thus it can't really be empty
+        Long beginningTimestamp = tuples.get(0).getLongByField("timestamp");
+        Long endTimestamp = tuples.get(tuples.size() - 1).getLongByField("timestamp");
+        int sumOfOperations = tuples.stream().mapToInt(tuple -> tuple.getIntegerByField("operation")).sum();
+        Values values = new Values(sumOfOperations, beginningTimestamp, endTimestamp);
+        outputCollector.emit(values);
+    }
+}

libraries-data/src/main/java/com/baeldung/storm/bolt/FileWritingBolt.java

@@ -0,0 +1,72 @@
+package com.baeldung.storm.bolt;
+
+import com.baeldung.storm.model.AggregatedWindow;
+import com.fasterxml.jackson.annotation.JsonAutoDetect;
+import com.fasterxml.jackson.annotation.PropertyAccessor;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.storm.task.OutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.topology.base.BaseRichBolt;
+import org.apache.storm.tuple.Tuple;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedWriter;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.util.Map;
+
+public class FileWritingBolt extends BaseRichBolt {
+    public static Logger logger = LoggerFactory.getLogger(FileWritingBolt.class);
+    private BufferedWriter writer;
+    private String filePath;
+    private ObjectMapper objectMapper;
+
+    @Override
+    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
+        objectMapper = new ObjectMapper();
+        objectMapper.setVisibility(PropertyAccessor.FIELD, JsonAutoDetect.Visibility.ANY);
+        try {
+            writer = new BufferedWriter(new FileWriter(filePath));
+        } catch (IOException e) {
+            logger.error("Failed to open a file for writing.", e);
+        }
+    }
+
+    @Override
+    public void execute(Tuple tuple) {
+        int sumOfOperations = tuple.getIntegerByField("sumOfOperations");
+        long beginningTimestamp = tuple.getLongByField("beginningTimestamp");
+        long endTimestamp = tuple.getLongByField("endTimestamp");
+
+        if(sumOfOperations > 200) {
+            AggregatedWindow aggregatedWindow = new AggregatedWindow(sumOfOperations, beginningTimestamp, endTimestamp);
+            try {
+                writer.write(objectMapper.writeValueAsString(aggregatedWindow));
+                writer.write("\n");
+                writer.flush();
+            } catch (IOException e) {
+                logger.error("Failed to write data to file.", e);
+            }
+        }
+    }
+
+    public FileWritingBolt(String filePath) {
+        this.filePath = filePath;
+    }
+
+    @Override
+    public void cleanup() {
+        try {
+            writer.close();
+        } catch (IOException e) {
+            logger.error("Failed to close the writer!");
+        }
+    }
+
+    @Override
+    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
+
+    }
+}

libraries-data/src/main/java/com/baeldung/storm/bolt/FilteringBolt.java

@@ -0,0 +1,22 @@
+package com.baeldung.storm.bolt;
+
+import org.apache.storm.topology.BasicOutputCollector;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.topology.base.BaseBasicBolt;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Tuple;
+
+public class FilteringBolt extends BaseBasicBolt {
+    @Override
+    public void execute(Tuple tuple, BasicOutputCollector basicOutputCollector) {
+        int operation = tuple.getIntegerByField("operation");
+        if(operation > 0 ) {
+            basicOutputCollector.emit(tuple.getValues());
+        }
+    }
+
+    @Override
+    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
+        outputFieldsDeclarer.declare(new Fields("operation", "timestamp"));
+    }
+}

libraries-data/src/main/java/com/baeldung/storm/bolt/PrintingBolt.java

@@ -0,0 +1,18 @@
+package com.baeldung.storm.bolt;
+
+import org.apache.storm.topology.BasicOutputCollector;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.topology.base.BaseBasicBolt;
+import org.apache.storm.tuple.Tuple;
+
+public class PrintingBolt extends BaseBasicBolt {
+    @Override
+    public void execute(Tuple tuple, BasicOutputCollector basicOutputCollector) {
+        System.out.println(tuple);
+    }
+
+    @Override
+    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
+
+    }
+}

libraries-data/src/main/java/com/baeldung/storm/model/AggregatedWindow.java

@@ -0,0 +1,16 @@
+package com.baeldung.storm.model;
+
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+
+@JsonSerialize
+public class AggregatedWindow {
+    int sumOfOperations;
+    long beginningTimestamp;
+    long endTimestamp;
+
+    public AggregatedWindow(int sumOfOperations, long beginningTimestamp, long endTimestamp) {
+        this.sumOfOperations = sumOfOperations;
+        this.beginningTimestamp = beginningTimestamp;
+        this.endTimestamp = endTimestamp;
+    }
+}

libraries-data/src/main/java/com/baeldung/storm/model/User.java

@@ -0,0 +1,40 @@
+package com.baeldung.storm.model;
+
+public class User {
+    private String username;
+    private String password;
+    private String email;
+    private int age;
+
+    public String getUsername() {
+        return username;
+    }
+
+    public void setUsername(String username) {
+        this.username = username;
+    }
+
+    public String getPassword() {
+        return password;
+    }
+
+    public void setPassword(String password) {
+        this.password = password;
+    }
+
+    public String getEmail() {
+        return email;
+    }
+
+    public void setEmail(String email) {
+        this.email = email;
+    }
+
+    public int getAge() {
+        return age;
+    }
+
+    public void setAge(int age) {
+        this.age = age;
+    }
+}

libraries-data/src/main/java/com/baeldung/storm/serialization/UserSerializer.java

@@ -0,0 +1,30 @@
+package com.baeldung.storm.serialization;
+
+
+import com.baeldung.storm.model.User;
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.Serializer;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+
+public class UserSerializer extends Serializer<User>{
+    @Override
+    public void write(Kryo kryo, Output output, User user) {
+        output.writeString(user.getEmail());
+        output.writeString(user.getUsername());
+        output.write(user.getAge());
+    }
+
+    @Override
+    public User read(Kryo kryo, Input input, Class<User> aClass) {
+        User user = new User();
+        String email = input.readString();
+        String name = input.readString();
+        int age = input.read();
+        user.setAge(age);
+        user.setEmail(email);
+        user.setUsername(name);
+
+        return  user;
+    }
+}

libraries-data/src/main/java/com/baeldung/storm/spout/RandomIntSpout.java

@@ -0,0 +1,35 @@
+package com.baeldung.storm.spout;
+
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.topology.base.BaseRichSpout;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Values;
+import org.apache.storm.utils.Utils;
+
+import java.util.Map;
+import java.util.Random;
+
+public class RandomIntSpout extends BaseRichSpout {
+
+    private Random random;
+    private SpoutOutputCollector outputCollector;
+
+    @Override
+    public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
+        random = new Random();
+        outputCollector = spoutOutputCollector;
+    }
+
+    @Override
+    public void nextTuple() {
+        Utils.sleep(1000);
+        outputCollector.emit(new Values(random.nextInt(), System.currentTimeMillis()));
+    }
+
+    @Override
+    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
+        outputFieldsDeclarer.declare(new Fields("randomInt", "timestamp"));
+    }
+}

libraries-data/src/main/java/com/baeldung/storm/spout/RandomNumberSpout.java

@@ -0,0 +1,40 @@
+package com.baeldung.storm.spout;
+
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.storm.task.OutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.topology.base.BaseRichSpout;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Values;
+import org.apache.storm.utils.Utils;
+
+import java.util.Map;
+import java.util.Random;
+
+public class RandomNumberSpout extends BaseRichSpout {
+    private Random random;
+    private SpoutOutputCollector collector;
+
+    @Override
+    public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
+        random = new Random();
+        collector = spoutOutputCollector;
+    }
+
+    @Override
+    public void nextTuple() {
+        Utils.sleep(1000);
+        //This will select random int from the range (0, 100)
+        int operation = random.nextInt(101);
+        long timestamp = System.currentTimeMillis();
+
+        Values values = new Values(operation, timestamp);
+        collector.emit(values);
+    }
+
+    @Override
+    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
+        outputFieldsDeclarer.declare(new Fields("operation", "timestamp"));
+    }
+}

libraries-data/src/main/resources/assembly/hadoop-job.xml

@@ -0,0 +1,28 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<assembly xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0"
+    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+    xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0 http://maven.apache.org/xsd/assembly-1.1.0.xsd">
+
+  <id>job</id>
+  <formats>
+    <format>jar</format>
+  </formats>
+  <includeBaseDirectory>false</includeBaseDirectory>
+  <dependencySets>
+    <dependencySet>
+      <unpack>false</unpack>
+      <scope>runtime</scope>
+      <outputDirectory>lib</outputDirectory>
+      <excludes>
+        <exclude>${groupId}:${artifactId}</exclude>
+      </excludes>
+    </dependencySet>
+    <dependencySet>
+      <unpack>true</unpack>
+      <includes>
+        <include>${groupId}:${artifactId}</include>
+      </includes>
+    </dependencySet>
+  </dependencySets>
+</assembly>

libraries-data/src/main/resources/kafka-connect/01_Quick_Start/connect-file-sink.properties

@@ -0,0 +1,20 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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.
+
+name=local-file-sink
+connector.class=FileStreamSink
+tasks.max=1
+file=test.sink.txt
+topics=connect-test

libraries-data/src/main/resources/kafka-connect/01_Quick_Start/connect-file-source.properties

@@ -0,0 +1,20 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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.
+
+name=local-file-source
+connector.class=FileStreamSource
+tasks.max=1
+file=test.txt
+topic=connect-test

libraries-data/src/main/resources/kafka-connect/01_Quick_Start/connect-standalone.properties

@@ -0,0 +1,44 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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.
+
+# These are defaults. This file just demonstrates how to override some settings.
+bootstrap.servers=localhost:9092
+
+# The converters specify the format of data in Kafka and how to translate it into Connect data. Every Connect user will
+# need to configure these based on the format they want their data in when loaded from or stored into Kafka
+key.converter=org.apache.kafka.connect.json.JsonConverter
+value.converter=org.apache.kafka.connect.json.JsonConverter
+# Converter-specific settings can be passed in by prefixing the Converter's setting with the converter we want to apply
+# it to
+key.converter.schemas.enable=false
+value.converter.schemas.enable=false
+
+offset.storage.file.filename=/tmp/connect.offsets
+# Flush much faster than normal, which is useful for testing/debugging
+offset.flush.interval.ms=10000
+
+# Set to a list of filesystem paths separated by commas (,) to enable class loading isolation for plugins
+# (connectors, converters, transformations). The list should consist of top level directories that include 
+# any combination of: 
+# a) directories immediately containing jars with plugins and their dependencies
+# b) uber-jars with plugins and their dependencies
+# c) directories immediately containing the package directory structure of classes of plugins and their dependencies
+# Note: symlinks will be followed to discover dependencies or plugins.
+# Examples: 
+# plugin.path=/usr/local/share/java,/usr/local/share/kafka/plugins,/opt/connectors,
+# Replace the relative path below with an absolute path if you are planning to start Kafka Connect from within a
+# directory other than the home directory of Confluent Platform.
+plugin.path=C:\Software\confluent-5.0.0\share\java
+#plugin.path=./share/java

libraries-data/src/main/resources/kafka-connect/02_Distributed/connect-distributed.properties

@@ -0,0 +1,88 @@
+##
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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.
+##
+
+# This file contains some of the configurations for the Kafka Connect distributed worker. This file is intended
+# to be used with the examples, and some settings may differ from those used in a production system, especially
+# the `bootstrap.servers` and those specifying replication factors.
+
+# A list of host/port pairs to use for establishing the initial connection to the Kafka cluster.
+bootstrap.servers=localhost:9092
+
+# unique name for the cluster, used in forming the Connect cluster group. Note that this must not conflict with consumer group IDs
+group.id=connect-cluster
+
+# The converters specify the format of data in Kafka and how to translate it into Connect data. Every Connect user will
+# need to configure these based on the format they want their data in when loaded from or stored into Kafka
+key.converter=org.apache.kafka.connect.json.JsonConverter
+value.converter=org.apache.kafka.connect.json.JsonConverter
+# Converter-specific settings can be passed in by prefixing the Converter's setting with the converter we want to apply
+# it to
+key.converter.schemas.enable=true
+value.converter.schemas.enable=true
+
+# Topic to use for storing offsets. This topic should have many partitions and be replicated and compacted.
+# Kafka Connect will attempt to create the topic automatically when needed, but you can always manually create
+# the topic before starting Kafka Connect if a specific topic configuration is needed.
+# Most users will want to use the built-in default replication factor of 3 or in some cases even specify a larger value.
+# Since this means there must be at least as many brokers as the maximum replication factor used, we'd like to be able
+# to run this example on a single-broker cluster and so here we instead set the replication factor to 1.
+offset.storage.topic=connect-offsets
+offset.storage.replication.factor=1
+#offset.storage.partitions=25
+
+# Topic to use for storing connector and task configurations; note that this should be a single partition, highly replicated,
+# and compacted topic. Kafka Connect will attempt to create the topic automatically when needed, but you can always manually create
+# the topic before starting Kafka Connect if a specific topic configuration is needed.
+# Most users will want to use the built-in default replication factor of 3 or in some cases even specify a larger value.
+# Since this means there must be at least as many brokers as the maximum replication factor used, we'd like to be able
+# to run this example on a single-broker cluster and so here we instead set the replication factor to 1.
+config.storage.topic=connect-configs
+config.storage.replication.factor=1
+
+# Topic to use for storing statuses. This topic can have multiple partitions and should be replicated and compacted.
+# Kafka Connect will attempt to create the topic automatically when needed, but you can always manually create
+# the topic before starting Kafka Connect if a specific topic configuration is needed.
+# Most users will want to use the built-in default replication factor of 3 or in some cases even specify a larger value.
+# Since this means there must be at least as many brokers as the maximum replication factor used, we'd like to be able
+# to run this example on a single-broker cluster and so here we instead set the replication factor to 1.
+status.storage.topic=connect-status
+status.storage.replication.factor=1
+#status.storage.partitions=5
+
+# Flush much faster than normal, which is useful for testing/debugging
+offset.flush.interval.ms=10000
+
+# These are provided to inform the user about the presence of the REST host and port configs 
+# Hostname & Port for the REST API to listen on. If this is set, it will bind to the interface used to listen to requests.
+#rest.host.name=
+#rest.port=8083
+
+# The Hostname & Port that will be given out to other workers to connect to i.e. URLs that are routable from other servers.
+#rest.advertised.host.name=
+#rest.advertised.port=
+
+# Set to a list of filesystem paths separated by commas (,) to enable class loading isolation for plugins
+# (connectors, converters, transformations). The list should consist of top level directories that include 
+# any combination of: 
+# a) directories immediately containing jars with plugins and their dependencies
+# b) uber-jars with plugins and their dependencies
+# c) directories immediately containing the package directory structure of classes of plugins and their dependencies
+# Examples: 
+# plugin.path=/usr/local/share/java,/usr/local/share/kafka/plugins,/opt/connectors,
+# Replace the relative path below with an absolute path if you are planning to start Kafka Connect from within a
+# directory other than the home directory of Confluent Platform.
+plugin.path=./share/java

libraries-data/src/main/resources/kafka-connect/02_Distributed/connect-file-sink.json

@@ -0,0 +1,9 @@
+{
+    "name": "local-file-sink",
+    "config": {
+        "connector.class": "FileStreamSink",
+        "tasks.max": 1,
+        "file": "test-distributed.sink.txt",
+        "topics": "connect-distributed"
+    }
+}

libraries-data/src/main/resources/kafka-connect/02_Distributed/connect-file-source.json

@@ -0,0 +1,9 @@
+{
+    "name": "local-file-source",
+    "config": {
+        "connector.class": "FileStreamSource",
+        "tasks.max": 1,
+        "file": "test-distributed.txt",
+        "topic": "connect-distributed"
+    }
+}

libraries-data/src/main/resources/kafka-connect/03_Transform/connect-distributed.properties

@@ -0,0 +1,88 @@
+##
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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.
+##
+
+# This file contains some of the configurations for the Kafka Connect distributed worker. This file is intended
+# to be used with the examples, and some settings may differ from those used in a production system, especially
+# the `bootstrap.servers` and those specifying replication factors.
+
+# A list of host/port pairs to use for establishing the initial connection to the Kafka cluster.
+bootstrap.servers=localhost:9092
+
+# unique name for the cluster, used in forming the Connect cluster group. Note that this must not conflict with consumer group IDs
+group.id=connect-cluster
+
+# The converters specify the format of data in Kafka and how to translate it into Connect data. Every Connect user will
+# need to configure these based on the format they want their data in when loaded from or stored into Kafka
+key.converter=org.apache.kafka.connect.json.JsonConverter
+value.converter=org.apache.kafka.connect.json.JsonConverter
+# Converter-specific settings can be passed in by prefixing the Converter's setting with the converter we want to apply
+# it to
+key.converter.schemas.enable=false
+value.converter.schemas.enable=false
+
+# Topic to use for storing offsets. This topic should have many partitions and be replicated and compacted.
+# Kafka Connect will attempt to create the topic automatically when needed, but you can always manually create
+# the topic before starting Kafka Connect if a specific topic configuration is needed.
+# Most users will want to use the built-in default replication factor of 3 or in some cases even specify a larger value.
+# Since this means there must be at least as many brokers as the maximum replication factor used, we'd like to be able
+# to run this example on a single-broker cluster and so here we instead set the replication factor to 1.
+offset.storage.topic=connect-offsets
+offset.storage.replication.factor=1
+#offset.storage.partitions=25
+
+# Topic to use for storing connector and task configurations; note that this should be a single partition, highly replicated,
+# and compacted topic. Kafka Connect will attempt to create the topic automatically when needed, but you can always manually create
+# the topic before starting Kafka Connect if a specific topic configuration is needed.
+# Most users will want to use the built-in default replication factor of 3 or in some cases even specify a larger value.
+# Since this means there must be at least as many brokers as the maximum replication factor used, we'd like to be able
+# to run this example on a single-broker cluster and so here we instead set the replication factor to 1.
+config.storage.topic=connect-configs
+config.storage.replication.factor=1
+
+# Topic to use for storing statuses. This topic can have multiple partitions and should be replicated and compacted.
+# Kafka Connect will attempt to create the topic automatically when needed, but you can always manually create
+# the topic before starting Kafka Connect if a specific topic configuration is needed.
+# Most users will want to use the built-in default replication factor of 3 or in some cases even specify a larger value.
+# Since this means there must be at least as many brokers as the maximum replication factor used, we'd like to be able
+# to run this example on a single-broker cluster and so here we instead set the replication factor to 1.
+status.storage.topic=connect-status
+status.storage.replication.factor=1
+#status.storage.partitions=5
+
+# Flush much faster than normal, which is useful for testing/debugging
+offset.flush.interval.ms=10000
+
+# These are provided to inform the user about the presence of the REST host and port configs 
+# Hostname & Port for the REST API to listen on. If this is set, it will bind to the interface used to listen to requests.
+#rest.host.name=
+#rest.port=8083
+
+# The Hostname & Port that will be given out to other workers to connect to i.e. URLs that are routable from other servers.
+#rest.advertised.host.name=
+#rest.advertised.port=
+
+# Set to a list of filesystem paths separated by commas (,) to enable class loading isolation for plugins
+# (connectors, converters, transformations). The list should consist of top level directories that include 
+# any combination of: 
+# a) directories immediately containing jars with plugins and their dependencies
+# b) uber-jars with plugins and their dependencies
+# c) directories immediately containing the package directory structure of classes of plugins and their dependencies
+# Examples: 
+# plugin.path=/usr/local/share/java,/usr/local/share/kafka/plugins,/opt/connectors,
+# Replace the relative path below with an absolute path if you are planning to start Kafka Connect from within a
+# directory other than the home directory of Confluent Platform.
+plugin.path=./share/java

libraries-data/src/main/resources/kafka-connect/03_Transform/connect-file-source-transform.json

@@ -0,0 +1,15 @@
+{
+    "name": "local-file-source",
+    "config": {
+        "connector.class": "FileStreamSource",
+        "tasks.max": 1,
+        "file": "transformation.txt",
+        "topic": "connect-transformation",
+        "transforms": "MakeMap,InsertSource",
+        "transforms.MakeMap.type": "org.apache.kafka.connect.transforms.HoistField$Value",
+        "transforms.MakeMap.field": "line",
+        "transforms.InsertSource.type": "org.apache.kafka.connect.transforms.InsertField$Value",
+        "transforms.InsertSource.static.field": "data_source",
+        "transforms.InsertSource.static.value": "test-file-source"
+    }
+}

libraries-data/src/main/resources/kafka-connect/04_Custom/connect-mongodb-sink.json

@@ -0,0 +1,14 @@
+{
+    "name": "mongodb-sink",
+    "config": {
+        "connector.class": "at.grahsl.kafka.connect.mongodb.MongoDbSinkConnector",
+        "tasks.max": 1,
+        "topics": "connect-custom",
+        "mongodb.connection.uri": "mongodb://mongo-db/test?retryWrites=true",
+        "mongodb.collection": "MyCollection",
+        "key.converter": "org.apache.kafka.connect.json.JsonConverter",
+        "key.converter.schemas.enable": false,
+        "value.converter": "org.apache.kafka.connect.json.JsonConverter",
+        "value.converter.schemas.enable": false
+    }
+}

libraries-data/src/main/resources/kafka-connect/04_Custom/connect-mqtt-source.json

@@ -0,0 +1,13 @@
+{
+    "name": "mqtt-source",
+    "config": {
+        "connector.class": "io.confluent.connect.mqtt.MqttSourceConnector",
+        "tasks.max": 1,
+        "mqtt.server.uri": "tcp://mosquitto:1883",
+        "mqtt.topics": "baeldung",
+        "kafka.topic": "connect-custom",
+        "value.converter": "org.apache.kafka.connect.converters.ByteArrayConverter",
+        "confluent.topic.bootstrap.servers": "kafka:9092",
+        "confluent.topic.replication.factor": 1
+    }
+}

libraries-data/src/main/resources/kafka-connect/04_Custom/docker-compose.yaml

@@ -0,0 +1,94 @@
+version: '3.3'
+
+services:
+  mosquitto:
+    image: eclipse-mosquitto:1.5.5
+    hostname: mosquitto
+    container_name: mosquitto
+    expose:
+      - "1883"
+    ports:
+      - "1883:1883"
+  zookeeper:
+    image: zookeeper:3.4.9
+    restart: unless-stopped
+    hostname: zookeeper
+    container_name: zookeeper
+    ports:
+      - "2181:2181"
+    environment:
+        ZOO_MY_ID: 1
+        ZOO_PORT: 2181
+        ZOO_SERVERS: server.1=zookeeper:2888:3888
+    volumes:
+      - ./zookeeper/data:/data
+      - ./zookeeper/datalog:/datalog
+  kafka:
+    image: confluentinc/cp-kafka:5.1.0
+    hostname: kafka
+    container_name: kafka
+    ports:
+      - "9092:9092"
+    environment:
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092
+      KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"
+      KAFKA_BROKER_ID: 1
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+    volumes:
+      - ./kafka/data:/var/lib/kafka/data
+    depends_on:
+      - zookeeper
+  kafka-connect:
+    image: confluentinc/cp-kafka-connect:5.1.0
+    hostname: kafka-connect
+    container_name: kafka-connect
+    ports:
+      - "8083:8083"
+    environment:
+      CONNECT_BOOTSTRAP_SERVERS: "kafka:9092"
+      CONNECT_REST_ADVERTISED_HOST_NAME: connect
+      CONNECT_REST_PORT: 8083
+      CONNECT_GROUP_ID: compose-connect-group
+      CONNECT_CONFIG_STORAGE_TOPIC: docker-connect-configs
+      CONNECT_OFFSET_STORAGE_TOPIC: docker-connect-offsets
+      CONNECT_STATUS_STORAGE_TOPIC: docker-connect-status
+      CONNECT_KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
+      CONNECT_VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
+      CONNECT_INTERNAL_KEY_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
+      CONNECT_INTERNAL_VALUE_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
+      CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: "1"
+      CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: "1"
+      CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: "1"
+      CONNECT_PLUGIN_PATH: '/usr/share/java,/etc/kafka-connect/jars'
+      CONNECT_CONFLUENT_TOPIC_REPLICATION_FACTOR: 1
+    volumes:
+      - /tmp/custom/jars:/etc/kafka-connect/jars
+    depends_on:
+      - zookeeper
+      - kafka
+      - mosquitto
+  mongo-db:
+    image: mongo:4.0.5
+    hostname: mongo-db
+    container_name: mongo-db
+    expose:
+      - "27017"
+    ports:
+      - "27017:27017"
+    command: --bind_ip_all --smallfiles
+    volumes:
+      - ./mongo-db:/data
+  mongoclient:
+    image: mongoclient/mongoclient:2.2.0
+    container_name: mongoclient
+    hostname: mongoclient
+    depends_on:
+      - mongo-db
+    ports:
+      - 3000:3000
+    environment:
+      MONGO_URL: "mongodb://mongo-db:27017"
+      PORT: 3000
+    expose:
+      - "3000"

libraries-data/src/main/resources/logback.xml

@@ -0,0 +1,15 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<configuration>
+    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n
+            </pattern>
+        </encoder>
+    </appender>
+    <logger name="io.ebean.DDL" level="TRACE"/>
+    <logger name="io.ebean.SQL" level="TRACE"/>
+    <logger name="io.ebean.TXN" level="TRACE"/>
+    <root level="WARN">
+        <appender-ref ref="STDOUT" />
+    </root>
+</configuration>