Maxime Beugnet

## Updates

The MongoDB Java quickstart repository is [available on Github](https://github.com/mongodb-developer/java-quick-start).

### March 25th, 2021

- Update Java Driver to 4.2.2.
- Added Client Side Field Level Encryption example.

### October 21th, 2020

- Update Java Driver to 4.1.1.
- The Java Driver logging is now enabled via the popular [SLF4J](http://www.slf4j.org/) API so I added logback in the `pom.xml` and a configuration file `logback.xml`.

## What's the Aggregation Pipeline?

<div>
 <img style="float: right;
 border-radius: 10px;
 margin-bottom: 30px;
 vertical-align: bottom;
 width: 30%;"
 src="https://www.mongodb.com/developer/images/qs-badges/qs-badge-java.png" alt="Java badge" />

The [aggregation pipeline](https://docs.mongodb.com/manual/core/aggregation-pipeline/) is a framework for data aggregation modeled on the concept of data processing pipelines, just like the "pipe" in the Linux Shell. Documents enter a multi-stage pipeline that transforms the documents into aggregated results.
</div>

It's the most powerful way to work with your data in MongoDB. It will allow us to make advanced queries like grouping documents, manipulate arrays, reshape document models, etc.

Let's see how we can harvest this power using Java.

## Getting Set Up

I will use the same repository as usual in this series. If you don't have a copy of it yet, you can clone it or just update it if you already have it:

``` sh
git clone https://github.com/mongodb-developer/java-quick-start
```

>If you didn't set up your free cluster on MongoDB Atlas, now is great time to do so. You have all the instructions in this [blog post](/quickstart/free-atlas-cluster/).

## First Example with Zips

In the [MongoDB Sample Dataset](https://docs.atlas.mongodb.com/sample-data/available-sample-datasets/) in [MongoDB Atlas](https://www.mongodb.com/cloud/atlas/), let's explore a bit the `zips` collection in the `sample_training` database.

``` javascript
MongoDB Enterprise Cluster0-shard-0:PRIMARY> db.zips.find({city:"NEW YORK"}).limit(2).pretty()
{
 "_id" : ObjectId("5c8eccc1caa187d17ca72f8a"),
 "city" : "NEW YORK",
 "zip" : "10001",
 "loc" : {
 "y" : 40.74838,
 "x" : 73.996705
 },
 "pop" : 18913,
 "state" : "NY"
}
{
 "_id" : ObjectId("5c8eccc1caa187d17ca72f8b"),
 "city" : "NEW YORK",
 "zip" : "10003",
 "loc" : {
 "y" : 40.731253,
 "x" : 73.989223
 },
 "pop" : 51224,
 "state" : "NY"
}
```

As you can see, we have one document for each zip code in the USA and for each, we have the associated population.

To calculate the population of New York, I would have to sum the population of each zip code to get the population of the entire city.

Let's try to find the 3 biggest cities in the state of Texas. Let's design this on paper first.

- I don't need to work with the entire collection. I need to filter only the cities in Texas.
- Once this is done, I can regroup all the zip code from a same city together to get the total population.
- Then I can order my cities by descending order or population.
- Finally I can keep the first 3 cities of my list.

The easiest way to build this pipeline in MongoDB is to use the [aggregation pipeline builder](https://docs.mongodb.com/compass/master/aggregation-pipeline-builder/) that is available in [MongoDB Compass](https://www.mongodb.com/products/compass) or in [MongoDB Atlas](https://www.mongodb.com/cloud/atlas/) in the `Collections` tab.

Once this is done, you can [export your pipeline to Java](https://docs.mongodb.com/compass/master/export-pipeline-to-language/) using the export button.

After a little code refactoring, here is what I have:

``` java
/**
 * find the 3 most densely populated cities in Texas.
 * @param zips sample_training.zips collection from the MongoDB Sample Dataset in MongoDB Atlas.
 */
private static void threeMostPopulatedCitiesInTexas(MongoCollection<Document> zips) {
 Bson match = match(eq("state", "TX"));
 Bson group = group("$city", sum("totalPop", "$pop"));
 Bson project = project(fields(excludeId(), include("totalPop"), computed("city", "$_id")));
 Bson sort = sort(descending("totalPop"));
 Bson limit = limit(3);

 List<Document> results = zips.aggregate(Arrays.asList(match, group, project, sort, limit))
 .into(new ArrayList<>());
 System.out.println("==> 3 most densely populated cities in Texas");
 results.forEach(printDocuments());
}
```

The MongoDB driver provides a lot of helpers to make the code easy to write and to read.

As you can see, I solved this problem with:

- A [$match stage](https://docs.mongodb.com/manual/reference/operator/aggregation/match/) to filter my documents and keep only the zip code in Texas,
- A [$group stage](https://docs.mongodb.com/manual/reference/operator/aggregation/group/) to regroup my zip codes in cities,
- A [$project stage](https://docs.mongodb.com/manual/reference/operator/aggregation/project/) to rename the field `_id` in `city` for a clean output (not mandatory but I'm classy),
- A [$sort stage](https://docs.mongodb.com/manual/reference/operator/aggregation/sort/) to sort by population descending,
- A [$limit stage](https://docs.mongodb.com/manual/reference/operator/aggregation/limit/) to keep only the 3 most populated cities.

Here is the output we get:

``` json
==> 3 most densely populated cities in Texas
{
 "totalPop": 2095918,
 "city": "HOUSTON"
}
{
 "totalPop": 940191,
 "city": "DALLAS"
}
{
 "totalPop": 811792,
 "city": "SAN ANTONIO"
}
```

In MongoDB 4.2, there are 30 different aggregation pipeline stages that you can use to manipulate your documents. If you want to know more, I encourage you to follow this course on MongoDB University: [M121: The MongoDB Aggregation Framework](https://university.mongodb.com/courses/M121/about).

## Second Example with Posts

This time, I'm using the collection `posts` in the same database.

``` json
MongoDB Enterprise Cluster0-shard-0:PRIMARY> db.posts.findOne()
{
 "_id" : ObjectId("50ab0f8bbcf1bfe2536dc3f9"),
 "body" : "Amendment I\nCongress shall make no law respecting an establishment of religion, or prohibiting the free exercise thereof; or abridging the freedom of speech, or of the press; or the right of the people peaceably to assemble, and to petition the Government for a redress of grievances.\n\nAmendment II\n\nA well regulated Militia, being necessary to the security of a free State, the right of the people to keep and bear Arms, shall not be infringed.\n\nAmendment III\n\nNo Soldier shall, in time of peace be quartered in any house, without the consent of the Owner, nor in time of war, but in a manner to be prescribed by law.\n\nAmendment IV\n\nThe right of the people to be secure in their persons, houses, papers, and effects, against unreasonable searches and seizures, shall not be violated, and no Warrants shall issue, but upon probable cause, supported by Oath or affirmation, and particularly describing the place to be searched, and the persons or things to be seized.\n\nAmendment V\n\nNo person shall be held to answer for a capital, or otherwise infamous crime, unless on a presentment or indictment of a Grand Jury, except in cases arising in the land or naval forces, or in the Militia, when in actual service in time of War or public danger; nor shall any person be subject for the same offence to be twice put in jeopardy of life or limb; nor shall be compelled in any criminal case to be a witness against himself, nor be deprived of life, liberty, or property, without due process of law; nor shall private property be taken for public use, without just compensation.\n\n\nAmendment VI\n\nIn all criminal prosecutions, the accused shall enjoy the right to a speedy and public trial, by an impartial jury of the State and district wherein the crime shall have been committed, which district shall have been previously ascertained by law, and to be informed of the nature and cause of the accusation; to be confronted with the witnesses against him; to have compulsory process for obtaining witnesses in his favor, and to have the Assistance of Counsel for his defence.\n\nAmendment VII\n\nIn Suits at common law, where the value in controversy shall exceed twenty dollars, the right of trial by jury shall be preserved, and no fact tried by a jury, shall be otherwise re-examined in any Court of the United States, than according to the rules of the common law.\n\nAmendment VIII\n\nExcessive bail shall not be required, nor excessive fines imposed, nor cruel and unusual punishments inflicted.\n\nAmendment IX\n\nThe enumeration in the Constitution, of certain rights, shall not be construed to deny or disparage others retained by the people.\n\nAmendment X\n\nThe powers not delegated to the United States by the Constitution, nor prohibited by it to the States, are reserved to the States respectively, or to the people.\"\n\n",
 "permalink" : "aRjNnLZkJkTyspAIoRGe",
 "author" : "machine",
 "title" : "Bill of Rights",
 "tags" : [
 "watchmaker",
 "santa",
 "xylophone",
 "math",
 "handsaw",
 "dream",
 "undershirt",
 "dolphin",
 "tanker",
 "action"
 ],
 "comments" : [
 {
 "body" : "Lorem ipsum dolor sit amet, consectetur adipisicing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua. Ut enim ad minim veniam, quis nostrud exercitation ullamco laboris nisi ut aliquip ex ea commodo consequat. Duis aute irure dolor in reprehenderit in voluptate velit esse cillum dolore eu fugiat nulla pariatur. Excepteur sint occaecat cupidatat non proident, sunt in culpa qui officia deserunt mollit anim id est laborum",
 "email" : "HvizfYVx@pKvLaagH.com",
 "author" : "Santiago Dollins"
 },
 {
 "body" : "Lorem ipsum dolor sit amet, consectetur adipisicing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua. Ut enim ad minim veniam, quis nostrud exercitation ullamco laboris nisi ut aliquip ex ea commodo consequat. Duis aute irure dolor in reprehenderit in voluptate velit esse cillum dolore eu fugiat nulla pariatur. Excepteur sint occaecat cupidatat non proident, sunt in culpa qui officia deserunt mollit anim id est laborum",
 "email" : "glbeRCMi@KwnNwhzl.com",
 "author" : "Omar Bowdoin"
 }
 ],
 "date" : ISODate("2012-11-20T05:05:15.231Z")
}
```

This collection of 500 posts has been generated artificially but it contains arrays and I want to show you how we can manipulate arrays in a pipeline.

Let's try to find the three most popular tags and for each tag, I also want the list of post titles they are tagging.

Here is my solution in Java.

``` java
/**
 * find the 3 most popular tags and their post titles
 * @param posts sample_training.posts collection from the MongoDB Sample Dataset in MongoDB Atlas.
 */
private static void threeMostPopularTags(MongoCollection<Document> posts) {
 Bson unwind = unwind("$tags");
 Bson group = group("$tags", sum("count", 1L), push("titles", "$title"));
 Bson sort = sort(descending("count"));
 Bson limit = limit(3);
 Bson project = project(fields(excludeId(), computed("tag", "$_id"), include("count", "titles")));
 List<Document> results = posts.aggregate(Arrays.asList(unwind, group, sort, limit, project)).into(new ArrayList<>());
 System.out.println("==> 3 most popular tags and their posts titles");
 results.forEach(printDocuments());
}
```

Here I'm using the very useful [$unwind stage](https://docs.mongodb.com/manual/reference/operator/aggregation/unwind/) to break down my array of tags.

It allows me in the following $group stage to group my tags, count the posts and collect the titles in a new array `titles`.

Here is the final output I get.

``` json
==> 3 most popular tags and their posts titles
{
 "count": 8,
 "titles": [
 "Gettysburg Address",
 "US Constitution",
 "Bill of Rights",
 "Gettysburg Address",
 "Gettysburg Address",
 "Declaration of Independence",
 "Bill of Rights",
 "Declaration of Independence"
 ],
 "tag": "toad"
}
{
 "count": 8,
 "titles": [
 "Bill of Rights",
 "Gettysburg Address",
 "Bill of Rights",
 "Bill of Rights",
 "Declaration of Independence",
 "Declaration of Independence",
 "Bill of Rights",
 "US Constitution"
 ],
 "tag": "forest"
}
{
 "count": 8,
 "titles": [
 "Bill of Rights",
 "Declaration of Independence",
 "Declaration of Independence",
 "Gettysburg Address",
 "US Constitution",
 "Bill of Rights",
 "US Constitution",
 "US Constitution"
 ],
 "tag": "hair"
}
```

As you can see, some titles are repeated. As I said earlier, the collection was generated so the post titles are not uniq. I could solve this "problem" by using the [$addToSet operator](https://docs.mongodb.com/manual/reference/operator/aggregation/addToSet/index.html) instead of the [$push](https://docs.mongodb.com/manual/reference/operator/aggregation/push/) one if this was really an issue.

## Final Code

``` java
package com.mongodb.quickstart;

import com.mongodb.client.MongoClient;
import com.mongodb.client.MongoClients;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoDatabase;
import org.bson.Document;
import org.bson.conversions.Bson;
import org.bson.json.JsonWriterSettings;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.function.Consumer;

import static com.mongodb.client.model.Accumulators.push;
import static com.mongodb.client.model.Accumulators.sum;
import static com.mongodb.client.model.Aggregates.*;
import static com.mongodb.client.model.Filters.eq;
import static com.mongodb.client.model.Projections.*;
import static com.mongodb.client.model.Sorts.descending;

public class AggregationFramework {

 public static void main(String[] args) {
 String connectionString = System.getProperty("mongodb.uri");
 try (MongoClient mongoClient = MongoClients.create(connectionString)) {
 MongoDatabase db = mongoClient.getDatabase("sample_training");
 MongoCollection<Document> zips = db.getCollection("zips");
 MongoCollection<Document> posts = db.getCollection("posts");
 threeMostPopulatedCitiesInTexas(zips);
 threeMostPopularTags(posts);
 }
 }

 /**
 * find the 3 most densely populated cities in Texas.
 * @param zips sample_training.zips collection from the MongoDB Sample Dataset in MongoDB Atlas.
 */
 private static void threeMostPopulatedCitiesInTexas(MongoCollection<Document> zips) {
 Bson match = match(eq("state", "TX"));
 Bson group = group("$city", sum("totalPop", "$pop"));
 Bson project = project(fields(excludeId(), include("totalPop"), computed("city", "$_id")));
 Bson sort = sort(descending("totalPop"));
 Bson limit = limit(3);

 List<Document> results = zips.aggregate(Arrays.asList(match, group, project, sort, limit))
 .into(new ArrayList<>());
 System.out.println("==> 3 most densely populated cities in Texas");
 results.forEach(printDocuments());
 }

 /**
 * find the 3 most popular tags and their post titles
 * @param posts sample_training.posts collection from the MongoDB Sample Dataset in MongoDB Atlas.
 */
 private static void threeMostPopularTags(MongoCollection<Document> posts) {
 Bson unwind = unwind("$tags");
 Bson group = group("$tags", sum("count", 1L), push("titles", "$title"));
 Bson sort = sort(descending("count"));
 Bson limit = limit(3);
 Bson project = project(fields(excludeId(), computed("tag", "$_id"), include("count", "titles")));

 List<Document> results = posts.aggregate(Arrays.asList(unwind, group, sort, limit, project)).into(new ArrayList<>());
 System.out.println("==> 3 most popular tags and their posts titles");
 results.forEach(printDocuments());
 }

 private static Consumer<Document> printDocuments() {
 return doc -> System.out.println(doc.toJson(JsonWriterSettings.builder().indent(true).build()));
 }
}
```

## Wrapping Up

The aggregation pipeline is very powerful. We have just scratched the surface with these two examples but trust me if I tell you that it's your best ally if you can master it.

>I encourage you to follow the [M121 course on MongoDB University](https://university.mongodb.com/courses/M121/about) to become an aggregation pipeline jedi.
>
>If you want to learn more and deepen your knowledge faster, I recommend you check out the M220J: MongoDB for Java Developers training available for free on [MongoDB University](https://university.mongodb.com/).

In the next blog post, I will explain to you the [Change Streams](https://docs.mongodb.com/manual/changeStreams/) in Java.

Learn how to use the Aggregation Pipeline using the MongoDB Java Driver.

Java - Aggregation Pipeline

## Updates

The MongoDB Java quickstart repository is [available on Github](https://github.com/mongodb-developer/java-quick-start).

### March 25th, 2021

- Update Java Driver to 4.2.2.
- Added Client Side Field Level Encryption example.

### October 21th, 2020

- Update Java Driver to 4.1.1.
- The Java Driver logging is now enabled via the popular [SLF4J](http://www.slf4j.org/) API so I added logback in the `pom.xml` and a configuration file `logback.xml`.

## Introduction

<div>
<img 
 style="float: right;
 border-radius: 10px;
 margin-bottom: 30px;
 vertical-align: bottom;
 width: 30%;"
 src="https://www.mongodb.com/developer/images/qs-badges/qs-badge-java.png" class="align-right" alt="Java badge" />

Java is an object-oriented programming language and MongoDB stores documents, which look a lot like objects. Indeed, this is not a coincidence because that's the core idea behind the MongoDB database.
</div>

In this blog post, as promised in the [first blog post](/quickstart/java-setup-crud-operations/) of this series, I will show you how to automatically map MongoDB documents to Plain Old Java Objects (POJOs) using only the MongoDB driver.

## Getting Set Up

I will use the same repository as usual in this series. If you don't have a copy of it yet, you can clone it or just update it if you already have it:

``` sh
git clone https://github.com/mongodb-developer/java-quick-start
```

If you haven't yet set up your free cluster on MongoDB Atlas, now is a great time to do so. You have all the instructions in this [blog post](/quickstart/free-atlas-cluster/).

## The Grades Collection

If you followed this series, you know that we have been working with the `grades` collection in the `sample_training` database. You can import it easily by [loading the sample dataset in MongoDB Atlas](https://docs.atlas.mongodb.com/sample-data/load-sample-data/).

Here is what a MongoDB document looks like in extended JSON format. I'm using the extended JSON because it's easier to identify the field types and we will need them to build the POJOs.

``` json
{
 "_id": {
 "$oid": "56d5f7eb604eb380b0d8d8ce"
 },
 "student_id": {
 "$numberDouble": "0"
 },
 "scores": [{
 "type": "exam",
 "score": {
 "$numberDouble": "78.40446309504266"
 }
 }, {
 "type": "quiz",
 "score": {
 "$numberDouble": "73.36224783231339"
 }
 }, {
 "type": "homework",
 "score": {
 "$numberDouble": "46.980982486720535"
 }
 }, {
 "type": "homework",
 "score": {
 "$numberDouble": "76.67556138656222"
 }
 }],
 "class_id": {
 "$numberDouble": "339"
 }
}
```

## POJOs

The first thing we need is a representation of this document in Java. For each document or subdocument, I need a corresponding POJO class.

As you can see in the document above, I have the main document itself and I have an array of subdocuments in the `scores` field. Thus, we will need 2 POJOs to represent this document in Java:

- One for the grade,
- One for the scores.

In the package `com.mongodb.quickstart.models`, I created two new POJOs: `Grade.java` and `Score.java`.

[Grade.java](https://github.com/mongodb-developer/java-quick-start/blob/master/src/main/java/com/mongodb/quickstart/models/Grade.java):

``` java
package com.mongodb.quickstart.models;

// imports

public class Grade {

 private ObjectId id;
 @BsonProperty(value = "student_id")
 private Double studentId;
 @BsonProperty(value = "class_id")
 private Double classId;
 private List<Score> scores;

 // getters and setters with builder pattern
 // toString()
 // equals()
 // hashCode()
}
```

>In the Grade class above, I'm using `@BsonProperty` to avoid violating Java naming conventions for variables, getters, and setters. This allows me to indicate to the mapper that I want the `"student_id"` field in JSON to be mapped to the `"studentId"` field in Java.

[Score.java](https://github.com/mongodb-developer/java-quick-start/blob/master/src/main/java/com/mongodb/quickstart/models/Score.java):

``` java
package com.mongodb.quickstart.models;

import java.util.Objects;

public class Score {

 private String type;
 private Double score;

 // getters and setters with builder pattern
 // toString()
 // equals()
 // hashCode()
}
```

As you can see, we took care of matching the Java types with the JSON value types to follow the same data model. You can read more about [types and documents in the documentation](https://mongodb.github.io/mongo-java-driver/3.12/bson/documents/).

## Mapping POJOs

Now that we have everything we need, we can start the MongoDB driver code.

I created a new class `MappingPOJO` in the `com.mongodb.quickstart` package and here are the key lines of code:

- I need a `ConnectionString` instance instead of the usual `String` I have used so far in this series. I'm still retrieving my MongoDB Atlas URI from the system properties. See my [starting and setup blog post](/quickstart/java-setup-crud-operations/) if you need a reminder.

``` java
ConnectionString connectionString = new ConnectionString(System.getProperty("mongodb.uri"));
```

- I need to configure the CodecRegistry to include a codec to handle the translation to and from BSON for our POJOs.

``` java
CodecRegistry pojoCodecRegistry = fromProviders(PojoCodecProvider.builder().automatic(true).build());
```

- And I need to add the default codec registry, which contains all the default codecs. They can handle all the major types in Java-like `Boolean`, `Double`, `String`, `BigDecimal`, etc.

``` java
CodecRegistry codecRegistry = fromRegistries(MongoClientSettings.getDefaultCodecRegistry(),
 pojoCodecRegistry);
```

- I can now wrap all my settings together using `MongoClientSettings`.

``` java
MongoClientSettings clientSettings = MongoClientSettings.builder()
 .applyConnectionString(connectionString)
 .codecRegistry(codecRegistry)
 .build();
```

- I can finally initialise my connection with MongoDB.

``` java
try (MongoClient mongoClient = MongoClients.create(clientSettings)) {
 MongoDatabase db = mongoClient.getDatabase("sample_training");
 MongoCollection<Grade> grades = db.getCollection("grades", Grade.class);
 [...]
}
```

As you can see in this last line of Java, all the magic is happening here. The `MongoCollection<Grade>` I'm retrieving is typed by `Grade` and not by `Document` as usual.

In the previous blog posts in this series, I showed you how to use CRUD operations by manipulating `MongoCollection<Document>`. Let's review all the CRUD operations using POJOs now.

- Here is an insert (create).

``` java
Grade newGrade = new Grade().setStudent_id(10003d)
 .setClass_id(10d)
 .setScores(singletonList(new Score().setType("homework").setScore(50d)));
grades.insertOne(newGrade);
```

- Here is a find (read).

``` java
Grade grade = grades.find(eq("student_id", 10003d)).first();
System.out.println("Grade found:\t" + grade);
```

- Here is an update with a `findOneAndReplace` returning the newest version of the document.

``` java
List<Score> newScores = new ArrayList<>(grade.getScores());
newScores.add(new Score().setType("exam").setScore(42d));
grade.setScores(newScores);
Document filterByGradeId = new Document("_id", grade.getId());
FindOneAndReplaceOptions returnDocAfterReplace = new FindOneAndReplaceOptions()
 .returnDocument(ReturnDocument.AFTER);
Grade updatedGrade = grades.findOneAndReplace(filterByGradeId, grade, returnDocAfterReplace);
System.out.println("Grade replaced:\t" + updatedGrade);
```

- And finally here is a delete.

``` java
System.out.println(grades.deleteOne(filterByGradeId));
```

## Final Code

`MappingPojo.java` ([code](https://github.com/mongodb-developer/java-quick-start/blob/master/src/main/java/com/mongodb/quickstart/MappingPOJO.java)):

``` java
package com.mongodb.quickstart;

import com.mongodb.ConnectionString;
import com.mongodb.MongoClientSettings;
import com.mongodb.client.MongoClient;
import com.mongodb.client.MongoClients;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoDatabase;
import com.mongodb.client.model.FindOneAndReplaceOptions;
import com.mongodb.client.model.ReturnDocument;
import com.mongodb.quickstart.models.Grade;
import com.mongodb.quickstart.models.Score;
import org.bson.Document;
import org.bson.codecs.configuration.CodecRegistry;
import org.bson.codecs.pojo.PojoCodecProvider;

import java.util.ArrayList;
import java.util.List;

import static com.mongodb.client.model.Filters.eq;
import static java.util.Collections.singletonList;
import static org.bson.codecs.configuration.CodecRegistries.fromProviders;
import static org.bson.codecs.configuration.CodecRegistries.fromRegistries;

public class MappingPOJO {

 public static void main(String[] args) {
 ConnectionString connectionString = new ConnectionString(System.getProperty("mongodb.uri"));
 CodecRegistry pojoCodecRegistry = fromProviders(PojoCodecProvider.builder().automatic(true).build());
 CodecRegistry codecRegistry = fromRegistries(MongoClientSettings.getDefaultCodecRegistry(), pojoCodecRegistry);
 MongoClientSettings clientSettings = MongoClientSettings.builder()
 .applyConnectionString(connectionString)
 .codecRegistry(codecRegistry)
 .build();
 try (MongoClient mongoClient = MongoClients.create(clientSettings)) {
 MongoDatabase db = mongoClient.getDatabase("sample_training");
 MongoCollection<Grade> grades = db.getCollection("grades", Grade.class);

 // create a new grade.
 Grade newGrade = new Grade().setStudent_id(10003d)
 .setClass_id(10d)
 .setScores(singletonList(new Score().setType("homework").setScore(50d)));
 grades.insertOne(newGrade);

 // find this grade.
 Grade grade = grades.find(eq("student_id", 10003d)).first();
 System.out.println("Grade found:\t" + grade);

 // update this grade: adding an exam grade
 List<Score> newScores = new ArrayList<>(grade.getScores());
 newScores.add(new Score().setType("exam").setScore(42d));
 grade.setScores(newScores);
 Document filterByGradeId = new Document("_id", grade.getId());
 FindOneAndReplaceOptions returnDocAfterReplace = new FindOneAndReplaceOptions().returnDocument(ReturnDocument.AFTER);
 Grade updatedGrade = grades.findOneAndReplace(filterByGradeId, grade, returnDocAfterReplace);
 System.out.println("Grade replaced:\t" + updatedGrade);

 // delete this grade
 System.out.println(grades.deleteOne(filterByGradeId));
 }
 }
}
```

To start this program, you can use this maven command line in your root project (where the `src` folder is) or your favorite IDE.

``` bash
mvn compile exec:java -Dexec.mainClass="com.mongodb.quickstart.MappingPOJO" -Dmongodb.uri="mongodb+srv://USERNAME:PASSWORD@cluster0-abcde.mongodb.net/test?w=majority"
```

## Wrapping Up

Mapping POJOs and your MongoDB documents simplifies your life a lot when you are solving real-world problems with Java, but you can certainly be successful without using POJOs.

MongoDB is a dynamic schema database which means your documents can have different schemas within a single collection. Mapping all the documents from such a collection can be a challenge. So, sometimes, using the "old school" method and the `Document` class will be easier.

>If you want to learn more and deepen your knowledge faster, I recommend you check out the M220J: MongoDB for Java Developers training available for free on [MongoDB University](https://university.mongodb.com/).

In the next blog post, I will show you the aggregation framework in Java.

Learn how to use the native mapping of POJOs using the MongoDB Java Driver.

Java - Mapping POJOs

Learn how to use the client side field level encryption using the MongoDB Java Driver.

Java - Client Side Field Level Encryption

Learn how to use MongoDB with Java in this tutorial on CRUD operations with example code and walkthrough.

Getting Started with MongoDB and Java - CRUD Operations Tutorial

Learn how to use the Change Streams using the MongoDB Java Driver.

Java - Change Streams

Mohit Sharma

> This article is targeted to existing Realm developers who want to understand how to migrate to Realm Kotlin SDK.

## Introduction

Android has changed a lot in recent years notably after the Kotlin language became a first-class 
citizen, so does the Realm SDK. Realm has recently moved its much-awaited Kotlin SDK to beta 
enabling developers to use Realm more fluently with Kotlin and opening doors to the world of Kotlin 
Multiplatform.

Let's understand the changes required when you migrate from Java to Kotlin SDK starting from setup
till its usage.

## Changes in setup

The new Realm Kotlin SDK is based on Kotlin Multiplatform architecture which enables you to have one
common module for all your data needs for all platforms. But this doesn't mean to use the new SDK
you would have to convert your existing Android app to KMM app right away, you can do that later.

Let's understand the changes needed in the gradle file to use Realm Kotlin SDK, by comparing the
previous implementation with the new one.

In project level `build.gradle`

Earlier with Java SDK

```kotlin
 buildscript {

 repositories {
 google()
 jcenter()
 }
 dependencies {
 classpath "com.android.tools.build:gradle:4.1.3"
 classpath "org.jetbrains.kotlin:kotlin-gradle-plugin:1.4.31"

 // Realm Plugin 
 classpath "io.realm:realm-gradle-plugin:10.10.1"

 // NOTE: Do not place your application dependencies here; they belong
 // in the individual module build.gradle files
 }
}
```

With Kotlin SDK, we can **delete the Realm plugin** from `dependencies`

```kotlin
 buildscript {

 repositories {
 google()
 jcenter()
 }
 dependencies {
 classpath "com.android.tools.build:gradle:4.1.3"
 classpath "org.jetbrains.kotlin:kotlin-gradle-plugin:1.4.31"

 // NOTE: Do not place your application dependencies here; they belong
 // in the individual module build.gradle files
 }
}
```

In the module-level `build.gradle`

With Java SDK, we 

1. Enabled Realm Plugin
2. Enabled Sync, if applicable

```groovy

plugins {
 id 'com.android.application'
 id 'kotlin-android'
 id 'kotlin-kapt'
 id 'realm-android'
}
```

```groovy
 android {

 ... ....

 realm {
 syncEnabled = true
 }
}
```

With Kotlin SDK,

1. Replace ``id 'realm-android'`` with ``id("io.realm.kotlin") version "0.10.0"``

```groovy
 plugins {
 id 'com.android.application'
 id 'kotlin-android'
 id 'kotlin-kapt'
 id("io.realm.kotlin") version "0.10.0"
}
```

2. Remove the Realm block under android tag

```groovy
 android {
 ... ....

 }
```

3. Add Realm dependency under `dependencies` tag

```groovy
 dependencies {

 implementation("io.realm.kotlin:library-sync:0.10.0")

 }
```

> If you are using only Realm local SDK, then you can add
> ```groovy
> dependencies {
> implementation("io.realm.kotlin:library-base:0.10.0")
> }
>```

With these changes, our Android app is ready to use Kotlin SDK.

## Changes in implementation

### Realm Initialization

Traditionally before using Realm for querying information in our project, we had to initialize and
set up few basic properties like name, version with sync config for database, let's update them 
as well.

Steps with JAVA SDK :

1. Call `Realm.init()`
2. Setup Realm DB properties like name, version, migration rules etc using `RealmConfiguration`.
3. Setup logging
4. Configure Realm Sync

With Kotlin SDK :

1. <s>Call `Realm.init()`</s> _Is not needed anymore_.
2. Setup Realm DB properties like db name, version, migration rules etc. using `RealmConfiguration`-
 _This remains the same apart from a few minor changes_.
3. Setup logging - _This is moved to `RealmConfiguration`_
4. Configure Realm Sync - _No changes_

### Changes to Models

No changes are required in model classes, except you might have to remove a few currently 
unsupported annotations like `@RealmClass` which is used for the embedded object.

> Note: You can remove `Open` keyword against `class` which was mandatory for using Java SDK in
> Kotlin.

### Changes to querying

The most exciting part starts from here 😎(IMO).

Traditionally Realm SDK has been on the top of the latest programming trends like Reactive
programming (Rx), LiveData and many more but with the technological shift in Android programming
language from Java to Kotlin, developers were not able to fully utilize the power of the language 
with Realm as underlying SDK was still in Java, few of the notable were support for the Coroutines, 
Kotlin Flow, etc.

But with the Kotlin SDK that all has changed and further led to the reduction of boiler code. 
Let's understand these by examples.

Example 1: As a user, I would like to register my visit as soon as I open the app or screen.

Steps to complete this operation would be

1. Authenticate with Realm SDK.
2. Based on the user information, create a sync config with the partition key.
3. Open Realm instance.
4. Start a Realm Transaction.
5. Query for current user visit count and based on that add/update count.

With JAVA SDK:

```kotlin
private fun updateData() {
 _isLoading.postValue(true)

 fun onUserSuccess(user: User) {
 val config = SyncConfiguration.Builder(user, user.id).build()

 Realm.getInstanceAsync(config, object : Realm.Callback() {
 override fun onSuccess(realm: Realm) {
 realm.executeTransactionAsync {
 var visitInfo = it.where(VisitInfo::class.java).findFirst()
 visitInfo = visitInfo?.updateCount() ?: VisitInfo().apply {
 partition = user.id
 }.updateCount()

 it.copyToRealmOrUpdate(visitInfo).apply {
 _visitInfo.postValue(it.copyFromRealm(this))
 }
 _isLoading.postValue(false)
 }
 }

 override fun onError(exception: Throwable) {
 super.onError(exception)
 // some error handling 
 _isLoading.postValue(false)
 }
 })
 }

 realmApp.loginAsync(Credentials.anonymous()) {
 if (it.isSuccess) {
 onUserSuccess(it.get())
 } else {
 _isLoading.postValue(false)
 }
 }
}
```

With Kotlin SDK:

```kotlin
private fun updateData() {
 viewModelScope.launch(Dispatchers.IO) {
 _isLoading.postValue(true)
 val user = realmApp.login(Credentials.anonymous())
 val config = SyncConfiguration.Builder(
 user = user,
 partitionValue = user.identity,
 schema = setOf(VisitInfo::class)
 ).build()

 val realm = Realm.open(configuration = config)
 realm.write {
 val visitInfo = this.query<VisitInfo>().first().find()
 copyToRealm(visitInfo?.updateCount()
 ?: VisitInfo().apply {
 partition = user.identity
 visitCount = 1
 })
 }
 _isLoading.postValue(false)
 }
}
```

Upon quick comparing, you would notice that lines of code have decreased by 30%, and we are using
coroutines for doing the async call, which is the natural way of doing asynchronous programming in
Kotlin. Let's check this with one more example.

Example 2: As user, I should be notified immediately about any change in user visit info. This is
more like observing the change to visit count.

With Java SDK:

```kotlin
 fun onRefreshCount() {
 _isLoading.postValue(true)

 fun getUpdatedCount(realm: Realm) {
 val visitInfo = realm.where(VisitInfo::class.java).findFirst()
 visitInfo?.let {
 _visitInfo.value = it
 _isLoading.postValue(false)
 }
 }

 fun onUserSuccess(user: User) {
 val config = SyncConfiguration.Builder(user, user.id).build()

 Realm.getInstanceAsync(config, object : Realm.Callback() {
 override fun onSuccess(realm: Realm) {
 getUpdatedCount(realm)
 }

 override fun onError(exception: Throwable) {
 super.onError(exception)
 //TODO: Implementation pending
 _isLoading.postValue(false)
 }
 })
 }

 realmApp.loginAsync(Credentials.anonymous()) {
 if (it.isSuccess) {
 onUserSuccess(it.get())
 } else {
 _isLoading.postValue(false)
 }
 }
}
```

With Kotlin SDK :

```kotlin

fun onRefreshCount(): Flow<VisitInfo?> {

 val user = runBlocking { realmApp.login(Credentials.anonymous()) }
 val config = SyncConfiguration.Builder(
 user = user,
 partitionValue = user.identity,
 schema = setOf(VisitInfo::class)
 ).build()

 val realm = Realm.open(config)
 return realm.query<VisitInfo>().first().asFlow()
}

```

Again upon quick comparing you would notice that lines of code have decreased drastically, by more
than **60%**, and apart coroutines for doing async call, we are using _Kotlin Flow_ to observe the
value changes.

With this, as mentioned earlier, we are further able to reduce our boilerplate code,
no [callback hell](http://callbackhell.com) and writing code is more natural now.

## Other major changes

Apart from Realm Kotlin SDK being written in Kotlin language, it is fundamentally little different 
from the JAVA SDK in a few ways:

- **Frozen by default**: All objects are now frozen. Unlike live objects, frozen objects do not
 automatically update after the database writes. You can still access live objects within a write
 transaction, but passing a live object out of a write transaction freezes the object.
- **Thread-safety**: All realm instances, objects, query results, and collections can now be
 transferred across threads.
- **Singleton**: You now only need one instance of each realm.

## Should you migrate now?

There is no straight answer to question, it really depends on usage, complexity of the app and time.
But I think so this the perfect time to evaluate the efforts and changes required to migrate as 
Realm Kotlin SDK would be the future.

This article is targeted to existing Realm developers who want to understand how to migrate to Realm Kotlin SDK.

How to migrate from Realm Java SDK to Realm Kotlin SDK

Pavel Duchovny

Data and event-driven applications are in high demand in a large variety of industries. With this demand, there is a growing challenge with how to sync the data across different data sources. 

A widely adopted solution for communicating real-time data transfer across multiple components in organization systems is implemented via clustered queues. One of the popular and proven solutions is [Apache Kafka](https://kafka.apache.org/).

The Kafka cluster is designed for streams of data that sequentially write events into commit logs, allowing real-time data movement between your services. Data is grouped into topics inside a Kafka cluster.

MongoDB provides a [Kafka connector](https://www.mongodb.com/kafka-connector) certified by [Confluent](https://www.confluent.io/), one of the largest Kafka providers. With the Kafka connector and Confluent software, you can publish data from a MongoDB cluster into Kafka topics using a source connector. Additionally, with a sink connector, you can consume data from a Kafka topic to persist directly and consistently into a MongoDB collection inside your MongoDB cluster.

In this article, we will provide a simple step-by-step guide on how to connect a remote Kafka cluster—in this case, a Confluent Cloud service—with a [MongoDB Atlas](https://mongodb.com/atlas) cluster. For simplicity purposes, the installation is minimal and designed for a small development environment. However, through the article, we will provide guidance and links for production-related considerations.

> **Pre-requisite**: To avoid JDK [known]((https://docs.mongodb.com/drivers/java/sync/current/faq/#how-do-i-fix---javax.net.ssl.sslhandshakeexception--extension--5--should-not-be-presented-in-certificate_request--)) certificate issues please update your JDK to one of the following patch versions or newer:
> - JDK 11.0.7+
> - JDK 13.0.3+
> - JDK 14.0.2+

## Table of Contents

1. [Create a Basic Confluent Cloud Cluster](#create-kafka-cluster)
1. [Create an Atlas Project and Cluster](#create-atlas-cluster) 
1. [Install Local Confluent Community Binaries to Run a Kafka Connect Instance](#install-kafka-connect)
1. [Configure the MongoDB Connector with Kafka Connect Locally](#configure-connectors)
1. [Start and Test Sink and Source MongoDB Kafka Connectors](#test-connectors)
1. [Summary](#summary)

## <a name="create-kafka-cluster"></a>Create a Basic Confluent Cloud Cluster

We will start by [creating](https://docs.confluent.io/cloud/current/clusters/create-cluster.html) a basic Kafka cluster in the [Confluent Cloud](https://confluent.cloud/). 

Once ready, create a topic to be used in the Kafka cluster. I created one named “orders.”

This “orders” topic will be used by Kafka Sink connector. Any data in this topic will be persisted automatically in the Atlas database.

![Kafka Cluster Topics](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/kafka_Topics_7eb45621e7.png)

You will also need another topic called "outsource.kafka.receipts". This topic will be used by the MongoDB Source connector, streaming reciepts from Atlas database.

[Generate](https://docs.confluent.io/cloud/current/access-management/authenticate/api-keys/api-keys.html) an `api-key` and `api-secret` to interact with this Kafka cluster. For the simplicity of this tutorial, I have selected the “Global Access” api-key. For production, it is recommended to give as minimum permissions as possible for the api-key used. Get a hold of the generated keys for future use.

Obtain the Kafka cluster connection string via `Cluster Overview > Cluster Settings > Identification > Bootstrap server` for future use. Basic clusters are open to the internet and in production, you will need to amend the access list for your specific hosts to connect to your cluster via advanced cluster ACLs.

## <a name="create-atlas-cluster"></a>Create a MongoDB Atlas Project and Cluster

[Create a project and cluster](https://docs.atlas.mongodb.com/getting-started/) or use an existing Atlas cluster in your project. 

![M0 Atlas Cluster](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/Screen_Shot_2021_12_13_at_14_04_39_5fc276d2d6.png)

Prepare your Atlas cluster for a kafka-connect [connection](https://docs.mongodb.com/kafka-connector/current/introduction/connect/). Inside your project’s access list, enable user and relevant IP addresses of your local host, the one used for Kafka Connect binaries. Finally, get a hold of the Atlas [connection string](https://docs.atlas.mongodb.com/connect-to-cluster/#use-the-connect-to-your-cluster-dialog-to-connect-to-your-cluster) for future use.

## <a name="install-kafka-connect"></a>Install a Kafka Connect Worker

[Kafka Connect](https://docs.confluent.io/platform/current/connect) is one of the mechanisms to reliably stream data between different data systems and a Kafka cluster. For production use, we recommend using a [distributed deployment](https://docs.confluent.io/home/connect/self-managed/userguide.html#standalone-vs-distributed-mode) for high availability, fault tolerance, and scalability. There is also a cloud version to [install the connector](https://docs.confluent.io/cloud/current/connectors/cc-mongo-db-sink.html) on the Confluent Cloud.

For this simple tutorial, we will use a standalone local Kafka Connect [installation](https://docs.confluent.io/platform/current/installation/installing_cp/zip-tar.html).

To have the binaries to install kafka-connect and all of its dependencies, let’s download the files:
```shell 
curl -O http://packages.confluent.io/archive/7.0/confluent-community-7.0.1.tar.gz
tar -xvf confluent-community-7.0.1.tar.gz
```

## <a name="configure-connectors"></a> Configure Kafka Connect

Configure the plugins directory where we will host the MongoDB Kafka Connector plugin:
```shell
mkdir -p /usr/local/share/kafka/plugins
```


Edit the `<confluent-install-dir>/etc/schema-registry/connect-avro-standalone.properties` using the content provided below. Ensure that you replace the `<kafka-cluster>:<kafka-port>` with information taken from Confluent Cloud bootstrap server earlier. 

Additionally, replace the generated `<api-key>` and `<api-secret>` taken from Confluent Cloud in every section.
```
bootstrap.servers=<kafka-cluster>:<kafka-port>

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 you want to apply
# it to
key.converter.schemas.enable=false
value.converter.schemas.enable=false

# The internal converter used for offsets and config data is configurable and must be specified, but most users will
# always want to use the built-in default. Offset and config data is never visible outside of Kafka Connect in this format.
internal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter=org.apache.kafka.connect.json.JsonConverter
internal.key.converter.schemas.enable=false
internal.value.converter.schemas.enable=false

# Store offsets on local filesystem
offset.storage.file.filename=/tmp/connect.offsets
# Flush much faster than normal, which is useful for testing/debugging
offset.flush.interval.ms=10000

ssl.endpoint.identification.algorithm=https


sasl.mechanism=PLAIN
request.timeout.ms=20000
retry.backoff.ms=500
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
username="<api-key>" password="<api-secret>";
security.protocol=SASL_SSL

consumer.ssl.endpoint.identification.algorithm=https
consumer.sasl.mechanism=PLAIN
consumer.request.timeout.ms=20000
consumer.retry.backoff.ms=500
consumer.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
username="<api-key>" password="<api-secret>";
consumer.security.protocol=SASL_SSL

producer.ssl.endpoint.identification.algorithm=https
producer.sasl.mechanism=PLAIN
producer.request.timeout.ms=20000
producer.retry.backoff.ms=500
producer.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
username="<api-key>" password="<api-secret>";
producer.security.protocol=SASL_SSL

plugin.path=/usr/local/share/kafka/plugins
```

**Important**: Place the `plugin.path` to point to our plugin directory with permissions to the user running the kafka-connect process.

### Install the MongoDB connector JAR: 
Download the “[all](https://docs.mongodb.com/kafka-connector/current/introduction/install/#install-the-connector-on-confluent-platform)” jar and place it inside the plugin directory.

```shell
cp ~/Downloads/mongo-kafka-connect-1.6.1-all.jar /usr/local/share/kafka/plugins/
```
### Configure a MongoDB Sink Connector

The [MongoDB Sink connector](https://docs.mongodb.com/kafka-connector/current/sink-connector/) will allow us to read data off a specific Kafka topic and write to a MongoDB collection inside our cluster. Create a MongoDB sink connector properties file in the main working dir: `mongo-sink.properties` with your Atlas cluster details replacing `<username>:<password>@<atlas-cluster>/<database>` from your Atlas connect tab. The working directory can be any directory that the `connect-standalone` binary has access to and its path can be provided to the `kafka-connect` command shown in ["Start Kafka Connect and Connectors"](#test-connectors) section.

```
name=mongo-sink
topics=orders
connector.class=com.mongodb.kafka.connect.MongoSinkConnector
tasks.max=1
connection.uri=mongodb+srv://<username>:<password>@<atlas-cluster>/<database>?retryWrites=true&w=majority
database=kafka
collection=orders
max.num.retries=1
retries.defer.timeout=5000
```

With the above configuration, we will listen to the topic called “orders” and publish the input documents into database `kafka` and collection name `orders`. 

### Configure Mongo Source Connector

The [MongoDB Source connector](https://docs.mongodb.com/kafka-connector/current/source-connector/) will allow us to read data off a specific MongoDB collection topic and write to a Kafka topic. When data will arrive into a collection called `receipts`, we can use a source connector to transfer it to a Kafka predefined topic named “outsource.kafka.receipts” (the configured prefix followed by the `<database>.<collection>` name as a topic—it's possible to use advanced mapping to change that). 

Let’s create file `mongo-source.properties` in the main working directory:
```
name=mongo-source
connector.class=com.mongodb.kafka.connect.MongoSourceConnector
tasks.max=1

# Connection and source configuration
connection.uri=mongodb+srv://<username>:<password>@<atlas-cluster>/<database>?retryWrites=true&w=majority
database=kafka
collection=receipts

topic.prefix=outsource
topic.suffix=
poll.max.batch.size=1000
poll.await.time.ms=5000

# Change stream options
pipeline=[]
batch.size=0
change.stream.full.document=updateLookup
publish.full.document.only=true
collation=
```

The main properties here are the database, collection, and aggregation pipeline used to listen for incoming changes as well as the connection string. The `topic.prefix` adds a prefix to the `<database>.<collection>` namespace as the Kafka topic on the Confluent side. In this case, the topic name that will receive new MongoDB records is “outsource.kafka.receipts” and was predefined earlier in this tutorial.

I have also added `publish.full.document.only=true` as I only need the actual document changed or inserted without the change stream event wrapping information.

### Start Kafka Connect and Connectors

For simplicity reasons, I am running the standalone Kafka Connect in the foreground.

```
 ./confluent-7.0.1/bin/connect-standalone ./confluent-7.0.1/etc/schema-registry/connect-avro-standalone.properties mongo-sink.properties mongo-source.properties
```

> **Important**: Run with the latest Java version to avoid JDK SSL bugs.

Now every document that will be populated to topic “orders” will be inserted into the `orders` collection using a sink connector. A source connector we configured will transmit every receipt document from `receipt` collection back to another topic called "outsource.kafka.receipts" to showcase a MongoDB consumption to a Kafka topic.

## <a name="test-connectors"></a> Publish Documents to the Kafka Queue

Through the Confluent UI, I have submitted a test document to my “orders” topic.
![Produce data into "orders" topic](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/Screen_Shot_2021_12_21_at_15_29_43_befe30feed.png)

### Atlas Cluster is Being Automatically Populated with the Data

Looking into my Atlas cluster, I can see a new collection named `orders` in the `kafka` database.

![Orders collection](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/Screen_Shot_2021_12_21_at_15_31_56_9a6f41a74d.png)

Now, let's assume that our application received the order document from the `orders` collection and produced a receipt. We can replicate this by inserting a document in the `kafka.reciepts` collection:

![Recipets collection](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/Screen_Shot_2021_12_21_at_15_33_22_0763e1d5a7.png)

This operation will cause the source connector to produce a message into “outsource.kafka.reciepts” topic.
### Kafka "outsource.kafka.reciepts" Topic

![Recieved data into "outsource.kafka.reciepts" topic](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/Screen_Shot_2021_12_21_at_15_33_56_7ab50fe570.png)

Log lines on kafka-connect will show that the process received and published the document: 

```
[2021-12-14 15:31:18,376] INFO [mongo-source|task-0] [Producer clientId=connector-producer-mongo-source-0] Cluster ID: lkc-65rmj (org.apache.kafka.clients.Metadata:287)
[2021-12-14 15:31:18,675] INFO [mongo-source|task-0] Opened connection [connectionId{localValue:21, serverValue:99712}] to dev-shard-00-02.uvwhr.mongodb.net:27017 (org.mongodb.driver.connection:71)
[2021-12-14 15:31:18,773] INFO [mongo-source|task-0] Started MongoDB source task (com.mongodb.kafka.connect.source.MongoSourceTask:203)
[2021-12-14 15:31:18,773] INFO [mongo-source|task-0] WorkerSourceTask{id=mongo-source-0} Source task finished initialization and start (org.apache.kafka.connect.runtime.WorkerSourceTask:233)
[2021-12-14 15:31:27,671] INFO [mongo-source|task-0|offsets] WorkerSourceTask{id=mongo-source-0} flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask:505
[2021-12-14 15:31:37,673] INFO [mongo-source|task-0|offsets] WorkerSourceTask{id=mongo-source-0} flushing 1 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask:505)
```

## <a name="summary"></a>Summary

In this how-to article, I have covered the fundamentals of building a simple yet powerful integration of MongoDB Atlas to Kafka clusters using MongoDB Kafka Connector and Kafka Connect.

This should be a good starting point to get you going with your next event-driven application stack and a successful integration between MongoDB and Kafka.

Try out [MongoDB Atlas](https://mongodb.com/atlas) and [Kafka connector ](https://www.mongodb.com/kafka-connector) today!

A simple step-by-step tutorial on how to use MongoDB Atlas with a Kafka Connector and connect it to any Remote Kafka Cluster.

Kafka to MongoDB Atlas End to End Tutorial

Igor Alekseev

Robert Walters

[Amazon Managed Streaming for Apache Kafka](https://aws.amazon.com/msk/features/) (MSK) is a fully managed, highly available Apache Kafka service. MSK makes it easy to ingest and process streaming data in real time and leverage that data easily within the AWS ecosystem. By being able to quickly stand up a Kafka solution, you spend less time managing infrastructure and more time solving your business problems, dramatically increasing productivity. MSK also supports integration of data sources such as MongoDB via the [AWS MSK Connect](https://aws.amazon.com/msk/features/msk-connect/) (Connect) service. This Connect service works with the [MongoDB Connector for Apache Kafka](https://www.mongodb.com/kafka-connector), enabling you to easily integrate MongoDB data.  

In this blog post, we will walk through how to set up MSK, configure the MongoDB Connector for Apache Kafka, and create a secured VPC Peered connection with MSK and a MongoDB Atlas cluster. The high-level process is as follows:

* Configure Amazon Managed Streaming for Apache Kafka
* Configure EC2 client 
* Configure a MongoDB Atlas Cluster
* Configure Atlas Private Link including VPC and subnet of the MSK
* Configure plugin in MSK for MongoDB Connector
* Create topic on MSK Cluster
* Install MongoSH command line tool on client
* Configure MongoDB Connector as a source or sink

In this example, we will have two collections in the same MongoDB cluster—the “source” and the “sink.” We will insert sample data into the source collection from the client, and this data will be consumed by MSK via the MongoDB Connector for Apache Kafka running as an MSK connector. As data arrives in the MSK topic, another instance of the MongoDB Connector for Apache Kafka will write the data to the MongoDB Atlas cluster “sink” collection. To align with best practices for secure configuration, we will set up an AWS Network Peered connection between the MongoDB Atlas cluster and the VPC containing MSK and the client EC2 instance.  

![Figure 1. Data flow between AWS MSK and MongoDB Atlas ](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/data_flow_mongodb_atlas_177d1bca15.png)

## Configure AWS Managed Service for Kafka

To create an Amazon MSK cluster using the AWS Management Console, sign in to the AWS Management Console, and open the [Amazon MSK console](https://console.aws.amazon.com/msk/home).

* Choose Create cluster and select Quick create.

For Cluster name, enter MongoDBMSKCluster.

For Apache Kafka version, select one that is 2.6.2 or above.

For broker type, select kafla.t3.small.

From the table under All cluster settings, copy the values of the following settings and save them because you will need them later in this blog:

* VPC
* Subnets
* Security groups associated with VPC

![Figure 2. Cluster settings showing VPC and Subnets](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/cluster_settings_95438aaef0.png)

* Choose “Create cluster.”

## Configure an EC2 client

Next, let's configure an EC2 instance to create a topic. This is where the MongoDB Atlas source collection will write to. This client can also be used to query the MSK topic and monitor the flow of messages from the source to the sink.

To create a client machine, open the Amazon EC2 console at [https://console.aws.amazon.com/ec2/](https://console.aws.amazon.com/ec2/).

* Choose Launch instances.
* Choose Select to create an instance of Amazon Linux 2 AMI (HVM) - Kernel 5.10, SSD Volume Type.
* Choose the t2.micro instance type by selecting the check box.
* Choose Next: Configure Instance Details.
* Navigate to the Network list and choose the VPC whose ID you saved in the previous step.
* Go to Auto-assign Public IP list and choose Enable.
* In the menu near the top, select Add Tags.
* Enter Name for the Key and MongoDBMSKCluster for the Value.
* Choose Review and Launch, and then click Launch.
* Choose Create a new key pair, enter MongoDBMSKKeyPair for Key pair name, and then choose Download Key Pair. Alternatively, you can use an existing key pair if you prefer.
* Start the new instance by pressing Launch Instances.

Next, we will need to configure the networking to allow connectivity between the client instance and the MSK cluster.

* Select View Instances. Then, in the Security Groups column, choose the security group that is associated with the MSKTutorialClient instance.
* Copy the name of the security group, and save it for later.
* Open the Amazon VPC console at [https://console.aws.amazon.com/vpc/](https://console.aws.amazon.com/vpc/).
* In the navigation pane, click on Security Groups. Find the security group whose ID you saved in Step 1 (Create an Amazon MSK Cluster). Choose this row by selecting the check box in the first column.
* In the Inbound Rules tab, choose Edit inbound rules.
* Choose Add rule.
* In the new rule, choose All traffic in the Type column. In the second field in the Source column, select the security group of the client machine. This is the group whose name you saved earlier in this step.
* Click Save rules.

The cluster's security group can now accept traffic that comes from the client machine's security group.

## Create MongoDB Atlas Cluster

To create a MongoDB Atlas Cluster, follow the Getting Started with Atlas tutorial. Note that in this blog, you will need to create an M30 Atlas cluster or above—as VPC peering is not available for M0, M2, and M5 clusters.

![Figure 3: Create new cluster dialog](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/create_new_cluster_144e85016b.png)

Once the cluster is created, configure an AWS private endpoint in the Atlas Network Access UI supplying the same subnets and VPC.  

* Click on Network Access.
* Click on Private Endpoint, and then the Add Private Endpoint button.

![Figure 4. Add private endpoint](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/creating_private_endpoint_ba8f2ce2b8.png)

* Fill out the VPC and subnet IDs from the previous section.

![Figure 5. Create private endpoint](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/create_private_endpoint_interface_dialogue_58645ad28f.png)

* SSH into the client machine created earlier and issue the following command in the Atlas portal: **aws ec2 create-vpc-endpoint **
    * Note that you may have to first configure the AWS CLI command using **aws configure** before you can create the VPC through this tool. See [Configuration Basics](https://docs.aws.amazon.com/cli/latest/userguide/cli-chap-configure.html) for more information.

## Configure MSK plugin

Next, we need to create a custom plugin for MSK. This custom plugin will be the MongoDB Connector for Apache Kafka. For reference, note that the connector will need to be uploaded to an S3 repository **before** you can create the plugin. You can download the MongoDB Connector for Apache Kafka from [Confluent Hub](https://www.confluent.io/hub/mongodb/kafka-connect-mongodb).

* Select “Create custom plugin” from the Custom Plugins menu within MSK.
* Fill out the custom plugin form, including the S3 location of the downloaded connector, and click “Create custom plugin.”

![Figure 6. Create custom plugin](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/create_custom_plugin_7514327ccf.png)

## Create topic on MSK cluster

When we start reading the data from MongoDB, we also need to create a topic in MSK to accept the data. On the client EC2 instance, let’s install Apache Kafka, which includes some basic tools. 

To begin, run the following command to install Java:

**sudo yum install java-1.8.0**

Next, run the command below to download Apache Kafka.

**wget https://archive.apache.org/dist/kafka/2.6.2/kafka_2.12-2.6.2.tgz**

Building off the previous step, run this command in the directory where you downloaded the TAR file:

**tar -xzf kafka_2.12-2.6.2.tgz**

The distribution of Kafka includes a **bin** folder with tools that can be used to manage topics. Go to the **kafka_2.12-2.6.2** directory.

To create the topic that will be used to write MongoDB events, issue this command:

`bin/kafka-topics.sh --create --zookeeper (INSERT YOUR ZOOKEEPER INFO HERE)--replication-factor 1 --partitions 1 --topic MongoDBMSKDemo.Source`

Also, remember that you can copy the Zookeeper server endpoint from the “View Client Information” page on your MSK Cluster. In this example, we are using plaintext.

![Figure 7. Apache ZooKeeper connection](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/apache_zookeeper_connection_6e23beafd6.png)

## Configure source connector

Once the plugin is created, we can create an instance of the MongoDB connector by selecting “Create connector” from the Connectors menu.

![Figure 8. Custom plugin dialog](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/choose_custom_plugin_f50bd58c30.png)

* Select the MongoDB plug in and click “Next.”
* Fill out the form as follows:

Conector name: **MongoDB Source Connector**

Cluster Type: **MSK Connector**

Select the MSK cluster that was created previously, and select “None” under the authentication drop down menu.

Enter your connector configuration (shown below) in the configuration settings text area.

`connector.class=com.mongodb.kafka.connect.MongoSourceConnector
database=MongoDBMSKDemo
collection=Source
tasks.max=1
connection.uri=(MONGODB CONNECTION STRING HERE)
value.converter=org.apache.kafka.connect.storage.StringConverter
key.converter=org.apache.kafka.connect.storage.StringConverter`

**Note**: You can find your Atlas connection string by clicking on the Connect button on your Atlas cluster. Select “Private Endpoint” if you have already configured the Private Endpoint above, then press “Choose a connection method.” Next, select “Connect your application” and copy the **mongodb+srv** connection string.

![Figure 9. Access permissions](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/access_permission_error_403a7a5e48.png)

In the “Access Permissions” section, you will need to create an IAM role with the [required trust policy](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-service-execution-role.html).

![Figure 10. Log delivery dialog](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/aws_msk_log_delivery_0a4ab236a0.png)

Once this is done, click “Next.” The last section will offer you the ability to use logs—which we highly recommend, as it will simplify the troubleshooting process. 

## Configure sink connector

Now that we have the source connector up and running, let’s configure a sink connector to complete the round trip. Create another instance of the MongoDB connector by selecting “Create connector” from the Connectors menu.

![Figure 8. Custom plugin dialog](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/choose_custom_plugin_f50bd58c30.png)

Select the same plugin that was created previously, and fill out the form as follows: 

Connector name: **MongoDB Sink Connector**
Cluster type: **MSK Connector**
Select the MSK cluster that was created previously and select “None” under the authentication drop down menu.
Enter your connector configuration (shown below) in the Configuration Settings text area.

`connector.class=com.mongodb.kafka.connect.MongoSinkConnector
database=MongoDBMSKDemo
collection=Sink
tasks.max=1
topics=MongoDBMSKDemo.Source
connection.uri=(MongoDB Atlas Connection String Gos Here)
value.converter=org.apache.kafka.connect.storage.StringConverter
key.converter=org.apache.kafka.connect.storage.StringConverter`

In the Access Permissions section, select the IAM role created earlier that has the [required trust policy](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-service-execution-role.html). As with the previous connector, be sure to leverage a log service like CloudWatch. 

Once the connector is successfully configured, we can test the round trip by writing to the Source collection and seeing the same data in the Sink collection.  

We can insert data in one of two ways: either through the intuitive Atlas UI, or with the new MongoSH (mongoshell) command line tool. Using [MongoSH](https://docs.mongodb.com/mongodb-shell/), you can interact directly with a MongoDB cluster to test queries, perform ad hoc database operations, and more. 

For your reference, we’ve added a section on how to use the mongoshell on your client EC2 instance below. 

## Install MongoDB shell on client

On the client EC2 instance, create a **/etc/yum.repos.d/mongodb-org-5.0.repo** file by typing:

`sudo nano /etc/yum.repos.d/mongodb-org-5.0.repo`

Paste in the following:

`[mongodb-org-5.0]
name=MongoDB Repository
baseurl=https://repo.mongodb.org/yum/amazon/2/mongodb-org/5.0/x86_64/
gpgcheck=1
enabled=1
gpgkey=[https://www.mongodb.org/static/pgp/server-5.0.asc](https://www.mongodb.org/static/pgp/server-5.0.asc)`

Next, install the MongoSH shell with this command:

`sudo yum install -y mongodb-mongosh`

Use the template below to connect to your MongoDB cluster via mongoshell: 

`mongosh “ (paste in your Atlas connection string here) “`

Once connected, type:
`Use MongoDBMSKDemo
db.Source.insertOne({“Testing”:123})`

To check the data on the sink collection, use this command:
`db.Sink.find({})`

If you run into any issues, be sure to check the log files. In this example, we used CloudWatch to read the events that were generated from MSK and the MongoDB Connector for Apache Kafka.

![Figure 12. Log events  ](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/cloud_watch_log_events_e5a0045c3d.png)

## Summary

Amazon Managed Streaming for Apache Kafka ([MSK](https://aws.amazon.com/msk/)) is a fully managed, secure, and highly available Apache Kafka service that makes it easy to ingest and process streaming data in real time. MSK allows you to import Kafka connectors such as the MongoDB Connector for Apache Kafka. These connectors make working with data sources seamless within MSK. In this article, you learned how to set up MSK, MSK Connect, and the MongoDB Connector for Apache Kafka.  You also learned how to set up a MongoDB Atlas cluster and configure it to use AWS network peering. To continue your learning, check out the following resources:

[MongoDB Connector for Apache Kafka Documentation](https://docs.mongodb.com/kafka-connector/current/)
[Amazon MSK Getting Started](https://docs.aws.amazon.com/msk/latest/developerguide/getting-started.html)
[Amazon MSK Connect Getting Started](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-getting-started.html)

In this article, learn how to set up Amazon MSK, configure the MongoDB Connector for Apache Kafka, and how it can be used as both a source and sink for data integration with MongoDB Atlas running in AWS.

Integrating MongoDB with Amazon Managed Streaming for Apache Kafka (MSK)

A high-level, class-based, object-oriented programming language. With Java Virtual Machine (JVM) Java applications are called WORA (Write Once Run Anywhere). 

Java

Featured

Getting Started with MongoDB and Java - CRUD Operations Tutorial

How to migrate from Realm Java SDK to Realm Kotlin SDK

Integrating MongoDB with Amazon Managed Streaming for Apache Kafka (MSK)

All Java Content