MongoDB Connector for Apache Kafka 1.7 Available Now

Robert Walters
February 17, 2022

Today, MongoDB has released version 1.7 of the MongoDB Connector for Apache Kafka! This article highlights some of the key features of this new release!

MongoDB errors to the Dead Letter Queue

Apache Kafka version 2.6 added support for handling errant records. The MongoDB Kafka Connector for Apache Kafka automatically sends messages that it cannot process to the dead letter queue. This includes messages that fail during conversion but up until this release did not include errors that were generated within MongoDB. For example, consider the scenario where we have a topic, “Sales.OrderStaging” This topic includes messages that contain an ‘order-id’ field. The application needs to insert a new document into MongoDB and use that order-id as the primary key or ‘_id’ of the document. If there happens to be a duplicate order-id entered on the kafka topic, the kafka message should be routed to a dead letter queue topic and the mongodb connector should continue to process other orders.

the following sink configuration highlights the configuration parameters that support this scenario:

"errors.tolerance":"all",
"mongo.errors.tolerance":"all",
"mongo.errors.log.enable":"true",
"errors.log.include.messages":"true",
"errors.deadletterqueue.topic.name":"orders.deadletterqueue",
"errors.deadletterqueue.context.headers.enable":"true",

"writemodel.strategy":"com.mongodb.kafka.connect.sink.writemodel.strategy.InsertOneDefaultStrategy",

"document.id.strategy":"com.mongodb.kafka.connect.sink.processor.id.strategy.PartialValueStrategy",
"document.id.strategy.overwrite.existing":"true",
"document.id.strategy.partial.value.projection.type": "AllowList",
"document.id.strategy.partial.value.projection.list": "order-id"

For example consider a kafka message with an order-id=5 and another message with the same order-id of 5. The sink connector will try to insert that second message with the same _id and there will be a MongoDB Error generated as expected. The kafka topic message which caused the error will be written to the orders.deadletterqueue topic. Once on the dead letter queue, you can inspect the errant records, update them, and resubmit them for processing. Setting the errors.deadletterqueue.context.headers.enable to true will add metadata to the DLQ message. This extra information may help with any automatic processing of errors in the queue. In addition to the DLQ, you can set errors.log.enable and error.log.include.messages configuration to write errors in the kafka connect log. Here is an example error from our scenario above:

com.mongodb.kafka.connect.sink.dlq.WriteException: v=1, code=11000, message=E11000 duplicate key error collection: Sales.Orders index: id dup key: { _id: { order-id: 5 } }, details={}

Bulk write improvements

Today the connector sink process works with a bulk insert in an ordered fashion. For example, consider these 10 documents in a bulk operation:

[1,2,3,4,5,6,7,8,9,10]

If document number 5 failed, perhaps due to a duplicate _id error, the MongoDB driver would return this error back to the connector and the rest of the documents would not be written to MongoDB e.g. only [1,2,3,4] is written in MongoDB in the above example. While this might be acceptable for some use cases, for other scenarios with large batch sizes this can make reprocessing messages cumbersome.

In Kafka 1.7, we introduced a new parameter bulk.write.ordered that by default is set to true which is the behavior as it exists today with the Kafka connector. Setting to false and running the above scenario will result in an end state of [1,2,3,4,6,7,8,9,10] written to MongoDB with 5 being written to the topic defined in the dead letter queue. Note that the actual order of the documents may be different since we specified false to bulk.write.ordered.

For more information on error handling including information on the format of the DLQ headers check out the MongoDB Kafka Connector documentation. To setup a dead letter queue, check out the Creating a dead letter queue section within the Confluent Kafka Connect documentation.

Changed retry logic

Currently the Kafka Connector manages retries of the writes to MongoDB using the max.num.retries and retries.defer.timeout configuration properties. This feature was originally intended to address challenges such as network connection issues. Since that time the MongoDB drivers have implemented native capabilities that handle retry logic: The Kafka Connector uses the MongoDB Java driver and has retries enabled by default so there are no changes or extra configuration you need to do to enable retry in the Kafka Connector. Note: If you set retryWrites to false in the connection.uri configuration property, then retries are disabled for the sink connector. If you would like to leverage the drivers native retry capability, simply remove the “retryWrites”' parameter from the connection.uri.

Allow disk use when copying

The copy.existing.allow.disk.use configuration copies existing data from the source. It uses an aggregation pipeline that filters change stream events coming from the MongoDB source. In certain situations this pipeline can use up large amounts of memory. This flag enabled by default allows the copy existing aggregation to use temporary disk storage if required by the query. Note the default is true, but set to false if the process running MongoDB doesn't have the permissions for disk access. For more information see the ‘allowDiskuse’ option in the aggregate() documentation.

← Previous

Deno x MongoDB: Check Out Our Latest Podcast

>> Announcement: Some features mentioned below will be deprecated on Sep. 30, 2025. Learn more . Used by the likes of Netflix, Uber, and PayPal, Node.js is one of today’s hottest technologies. Its versatility, scalability, and event-handling abilities quickly earned applause (and heavy use) from developers worldwide. Created by Ryan Dahl in 2008, Node.js solves several problems simultaneously. It enables programmers to use Javascript on the backend and the frontend, removing the need for workarounds. Additionally, its single-threaded architecture allows for a lightweight, rapid framework that can easily handle real-time events — critical for any app in an increasingly connected world. Recently, Dahl released Deno, another Javascript-based framework that improves on Node.js in areas like speed, efficiency, ease of use, testing, and packaging. Intrigued by this promising new product, Michael Lynn, Principal Developer Advocate at MongoDB, sat down with Jesse Hall, Senior Developer Advocate at MongoDB, to talk all things Deno, including strengths, structure, scenarios for use, and much more. Below are some highlights from their discussion. To listen to the full episode, check out the MongoDB Podcast here . The successor to Node.js Michael starts off the podcast with the question we’ve all been waiting for: What’s so special about Deno? After all, Deno is a runtime for Javascript and Typescript — similar to Node.js, at least superficially. “Why would we want to use Deno over a standard Node.js implementation?” Michael asks. “By default it’s secure,” Jesse replies, especially in contrast to Node.js. “That is one of the big selling factors, because you have to specifically tell it to open your network in order to allow access to read or write files.” In essence, Jesse explains, Deno is “set up to be the successor to Node.js.” As such, it includes a number of helpful features and attributes, such as blazing-fast speed (thanks to its Rust-based architecture), support for ES modules, and built-in asynchronous functions (removing the need to create your own). Better package handling = increased efficiency Deno also streamlines the development process in various ways — particularly when it concerns packages. Unlike Node, which requires a NPM package manager, Deno simply “downloads [packages] globally and caches them,” Jesse says, removing the need for a Node modules folder in your project. Because Deno pulls packages in through URLs, this removes the bloat associated with having a Node modules folder in a project. Michael wants to know if this affects development negatively — especially when it comes to offline development. Even if the files are located locally, surely Deno must require some sort of cached folder or module structure? That’s not necessarily the case, Jesse replies. “You do have a local cache of those modules, and when you actually import them in the URL string, you can specify what version (if you wanted a different version).” Deno x MongoDB Of course, given that both Michael and Jesse work at MongoDB, the million-dollar question remains: compatibility. “What does the support for MongoDB look like in Deno?” Michael asks. For now, Jesse explains, support comes in the form of third-party, community-supported modules only — specifically a Deno module called Mongo, easily accessible through the Deno website. As an alternative, Jesse suggests the MongoDB Atlas Data API , “which will allow you to bring your MongoDB data into basically anything, because it doesn’t require drivers at all.” And much more... For more detail, listen to the full podcast here . For a detailed walkthrough on using the Atlas Data API to import your MongoDB data into Deno, check out Jesse’s article here .

February 17, 2022

Next →

AI-Powered Call Centers: A New Era of Customer Service

Customer satisfaction is critical for insurance companies. Studies have shown that companies with superior customer experiences consistently outperform their peers. In fact, McKinsey found that life and property/casualty insurers with superior customer experiences saw a significant 20% and 65% increase in Total Shareholder Return , respectively, over five years. A satisfied customer is a loyal customer. They are 80% more likely to renew their policies, directly contributing to sustainable growth. However, one major challenge faced by many insurance companies is the inefficiency of their call centers. Agents often struggle to quickly locate and deliver accurate information to customers, leading to frustration and dissatisfaction. This article explores how Dataworkz and MongoDB can transform call center operations. By converting call recordings into searchable vectors (numerical representations of data points in a multi-dimensional space), businesses can quickly access relevant information and improve customer service. We'll dig into how the integration of Amazon Transcribe, Cohere, and MongoDB Atlas Vector Search—as well as Dataworkz's RAG-as-a-service platform— is achieving this transformation. From call recordings to vectors: A data-driven approach Customer service interactions are goldmines of valuable insights. By analyzing call recordings, we can identify successful resolution strategies and uncover frequently asked questions. In turn, by making this information—which is often buried in audio files— accessible to agents, they can give customers faster and more accurate assistance. However, the vast volume and unstructured nature of these audio files make it challenging to extract actionable information efficiently. To address this challenge, we propose a pipeline that leverages AI and analytics to transform raw audio recordings into vectors as shown in Figure 1: Storage of raw audio files: Past call recordings are stored in their original audio format Processing of the audio files with AI and analytics services (such as Amazon Transcribe Call Analytics ): speech-to-text conversion, summarization of content, and vectorization Storage of vectors and metadata: The generated vectors and associated metadata (e.g., call timestamps, agent information) are stored in an operational data store Figure 1: Customer service call insight extraction and vectorization flow Once the data is stored in vector format within the operational data store, it becomes accessible for real-time applications. This data can be consumed directly through vector search or integrated into a retrieval-augmented generation (RAG) architecture, a technique that combines the capabilities of large language models (LLMs) with external knowledge sources to generate more accurate and informative outputs. Introducing Dataworkz: Simplifying RAG implementation Building RAG pipelines can be cumbersome and time-consuming for developers who must learn yet another stack of technologies. Especially in this initial phase, where companies want to experiment and move fast, it is essential to leverage tools that allow us to abstract complexity and don’t require deep knowledge of each component in order to experiment with and realize the benefits of RAG quickly. Dataworkz offers a powerful and composable RAG-as-a-service platform that streamlines the process of building RAG applications for enterprises. To operationalize RAG effectively, organizations need to master five key capabilities: ETL for LLMs: Dataworkz connects with diverse data sources and formats, transforming the data to make it ready for consumption by generative AI applications. Indexing: The platform breaks down data into smaller chunks and creates embeddings that capture semantics, storing them in a vector database. Retrieval: Dataworkz ensures the retrieval of accurate information in response to user queries, a critical part of the RAG process. Synthesis: The retrieved information is then used to build the context for a foundational model, generating responses grounded in reality. Monitoring: With many moving parts in the RAG system, Dataworkz provides robust monitoring capabilities essential for production use cases. Dataworkz's intuitive point-and-click interface (as seen in Video 1) simplifies RAG implementation, allowing enterprises to quickly operationalize AI applications. The platform offers flexibility and choice in data connectors, embedding models, vector stores, and language models. Additionally, tools like A/B testing ensure the quality and reliability of generated responses. This combination of ease of use, optionality, and quality assurance is a key tenet of Dataworkz's "RAG as a Service" offering. Diving deeper: System architecture and functionalities Now that we’ve looked at the components of the pre-processing pipeline, let’s explore the proposed real-time system architecture in detail. It comprises the following modules and functions (see Figure 2): Amazon Transcribe , which receives the audio coming from the customer’s phone and converts it into text. Cohere ’s embedding model, served through Amazon Bedrock , vectorizes the text coming from Transcribe. MongoDB Atlas Vector Search receives the query vector and returns a document that contains the most semantically similar FAQ in the database. Figure 2: System architecture and modules Here are a couple of FAQs we used for the demo: Q: “Can you explain the different types of coverage available for my home insurance?” A: “Home insurance typically includes coverage for the structure of your home, your personal belongings, liability protection, and additional living expenses in case you need to temporarily relocate. I can provide more detailed information on each type if you'd like.” Q: “What is the process for adding a new driver to my auto insurance policy?" A: “To add a new driver to your auto insurance policy, I'll need some details about the driver, such as their name, date of birth, and driver's license number. We can add them to your policy over the phone, or you can do it through our online portal.” Note that the question is reported just for reference, and it’s not used for retrieval. The actual question is provided by the user through the voice interface and then matched in real-time with the answers in the database using Vector Search. This information is finally presented to the customer service operator in text form (see Fig. 3). The proposed architecture is simple but very powerful, easy to implement, and effective. Moreover, it can serve as a foundation for more advanced use cases that require complex interactions, such as agentic workflows , and iterative and multi-step processes that combine LLMs and hybrid search to complete sophisticated tasks. Figure 3: App interface, displaying what has been asked by the customer (left) and how the information is presented to the customer service operator (right) This solution not only impacts human operator workflows but can also underpin chatbots and voicebots, enabling them to provide more relevant and contextual customer responses. Building a better future for customer service By seamlessly integrating analytical and operational data streams, insurance companies can significantly enhance both operational efficiency and customer satisfaction. Our system empowers businesses to optimize staffing, accelerate inquiry resolution, and deliver superior customer service through data-driven, real-time insights. To embark on your own customer service transformation, explore our GitHub repository and take advantage of the Dataworkz free tier .

November 27, 2024