To use Hazelcast Jet Enterprise, you must set the license key using one of the configuration methods shown below. You can request a trial license key at hazelcast.com.

Hazelcast Jet Enterprise license keys are required only to run the Jet cluster. A Jet client can access the Enterprise features without the license key.

The license key can be configured using one of the following methods:

Hazelcast Jet Management Center is a management and monitoring suite providing a live overview of the Hazelcast Jet cluster. It’s a standalone tool with a web console.

Please see the Hazelcast Jet Management Center Reference Manual for the installation instructions.

Here’s a simple example of list transformation with the Stream API:

Here’s the equivalent in Jet. Note that we’re transforming Hazelcast IList s:

Here’s an example of grouping and aggregation with the Stream API. We compute a histogram of words by their length:

And here’s how to aggregate in Jet:

Note that the result of aggregate is just another pipeline stage, you can apply more transforms to it before draining to the sink.

If you have ever written your own Collector for the Stream API, you’ll find that Jet’s AggregateOperation is quite similar and you can transfer your skill to it.

Here’s a Stream API collector that computes the sum of input items:

And here’s Jet’s aggregate operation doing the same:

Compared to Collector, AggregateOperation defines two more primitives:

To gain more control over how Jet will run your job, you can pass in a JobConfig instance. For example, you can give your job a human-readable name:

Please note that you can have a single active job, i.e., running / suspended / waiting to be run, in the cluster with the same name. You get JobAlreadyExistsException if you make another submission. You can reuse the job name after the active job is completed.

If you have a microservice deployment where each package contains a jet member and makes the same job submission, you can ensure that the job runs only once by using the newJobIfAbsent() API. If a named job is submitted with this API, Jet does not create a second job if there is already an active job with the same name. Instead, it returns you a reference for the active job, as shown in the following snippet.

The API to submit a job to Jet is in a way deceptively simple: "just call a method". As long as you’re toying around with Jet instances started locally in a single JVM, everything will indeed work. However, as soon as you try to deploy to an actual cluster, you’ll face the consequences of the fact that your job definition must travel over the wire to reach remote members which don’t have your code on their classpath.

Your custom code must be packaged with the Jet job. For simple examples you can have everything in a single class and use code like this:

If you forget to do this, or don’t add all the classes involved, you may get a quite confusing exception:

SerializedLambda actually declares readResolve(), which would normally transform it into an instance of the correct functional interface type. If this method throws an exception, Java doesn’t report it but keeps the SerializedLambda instance and continues the deserialization. Later in the process it will try to assign it to a field whose type is the target type of the lambda (FunctionEx in the example above) and at that point it will fail with the ClassCastException. If you see this kind of error, double-check the list of classes you have added to the Jet job.

For more complex jobs it will become more practical to first package the job in a JAR and then use a command-line utility to submit it, as explained next.

Jet comes with the jet.sh script, which allows you to submit a Jet job packaged in a JAR file. You can find it in the Jet distribution zip file, in the bin directory. On Windows use jet.bat. To use it, follow these steps:

For example, write a class like this:

After building the JAR, submit the job:

jet.newJob() and jet.getJob(jobId) return a Job object, which you can use to monitor the job and change its status:

You can also get the job’s name, configuration, and submission time via job.getName(), job.getConfig(), and job.getSubmissionTime(). job.getStatus() will give you the current status of the job (running, failed, completed etc.). You can also call job.getFuture() to block until the job completes and then get its final outcome (either success or failure).

Jet does not support canceling the job with future.cancel(), instead you must call job.cancel(). This is due to the mismatch in the semantics between future.cancel() on one side and job.cancel() plus job.getStatus() on the other: the future immediately transitions to “completed by cancellation”, but it will take some time until the actual job in the cluster changes to that state. Not to confuse the users with these differences we decided to make future.cancel() fail with an exception.

Note: Job Upgrade is only available in Jet Enterprise

Jet allows you to upgrade a job while preserving its state. This is useful for bug fixing or to improve the application.

The new pipeline has to be compatible with the old one, for more details about what you can change in the pipeline, see Update a DAG Without Losing the State.

Job Upgrades use the snapshots. The Job state is exported to the snapshot and the new job version starts from from the snapshot.

If you want to preserve the state of the cancelled job, use cancelAndExportSnapshot:

You can also export the state and keep the job running. Both old and new version can then run in parallel. This is useful e.g. for an A/B testing.

Then use the exported snapshot to submit a new job. The Job will use the state in the snapshot as its initial state:

The exported snapshots used for Job Upgrades differ from the snapshots used for the Fault Tolerance. Exported snapshots aren’t deleted automatically by Jet. Therefore you can also use the exported snapshot as a point of recovery. When you no longer need the exported snapshot, delete it.

Jet keeps an inventory of all the jobs submitted to it, including those that have already completed. Access the full list with jet.getJobs(). You can use any Job instance from that list to monitor and manage a job, whether it was you or some other client that submitted it.

This example tells you what Jet has been up to in the last five minutes:

To only return all jobs submitted with a particular name, you can call jet.getJobs(name), or jet.getJob(name) to get just the latest one.

Here’s how you can check the statistics on a job named my-job:

Note: data about completed jobs are evicted after 7 days.

"Exactly-once processing" means the output is consistent with processing each stream item exactly once. This is the ultimate guarantee you can ask for.

As of version 0.6, Hazelcast Jet supports exactly-once processing with the source being either a Hazelcast IMap journal or a Kafka topic, and the sink being a Hazelcast IMap.

If you configure Jet for exactly-once but use Kafka as the sink, after a job restart you may get duplicates in the output. As opposed to doubly processing an input item, this is more benign because it just means getting the exact same result twice.

"At-least-once processing" means the output is consistent with processing each stream item at least once. Some items may get processed again after a restart, as if they represented new events. This is a lesser guarantee that Jet can deliver at a higher throughput and lower latency. In some cases it is good enough.

In some other cases, however, duplicate processing of data items can have quite surprising consequences. There is more information about this in our Jet Concepts chapter.

Fault tolerance is off by default. To activate it for a job, create a JobConfig object and set the processing guarantee. You can also configure snapshot interval.

Using less frequent snapshots, more data will have to be replayed and the temporary spike in the latency of the output will be greater. More frequent snapshots will reduce the throughput and introduce more latency variation during regular processing.

You can configure the behavior of what will happen when members are added or removed from the cluster.

By default, auto-scaling is enabled.

Jet doesn’t delegate its fault tolerance to an outside system, it backs up the state to its own IMap objects. IMap is a replicated in-memory data structure, storing each key-value pair on a configurable number of cluster members. By default it will make a single backup copy, resulting in a system that tolerates the failure of a single member at a time. You can tweak this setting when starting Jet, for example increase the backup count to two:

Note: if multiple members are lost simultaneously, some data from the backing IMaps can be lost. This is not currently checked and the job will restart with some state data from the snapshot missing, or it might fail if classpath resources were added and are missing. We plan to address this in future releases.

A specific kind of failure is a so-called "split brain". It happens on network failure when a member or members think the other members left the cluster, but in fact they still run, but don’t see each other over the network. Now we have two or more fully-functioning Jet clusters where there was supposed to be one. Each one will recover and restart the same Jet job, making it to run multiple times.

Hazelcast Jet offers a mechanism to reduce this hazard: split-brain protection. It works by ensuring that a job can be restarted only in a cluster whose size is more than half of what it was before the job was suspended. Enable split-brain protection like this:

If there’s an even number of members in your cluster, this may mean the job will not be able to restart at all if the cluster splits into two equally-sized parts. We recommend having an odd number of members.

Note also that you should ensure there is no split-brain condition at the moment you are introducing new members to the cluster. If that happens, both sub-clusters may grow to more than half of the previous size. This will defuse the split-brain protection mechanism.

When it comes to serializing the description of a Jet job, performance is not critical. However, for the data passing through the pipeline, the cost of the serialize-deserialize cycle can easily dwarf the cost of actual data transfer, especially on high-end LANs typical for data centers. In this context the performance of Java serialization is so poor that it regularly becomes the bottleneck. This is due to its heavy usage of reflection, overheads in the serialized form, etc.

Since Hazelcast IMDG faced the same problem a long time ago, we have mature support for optimized custom serialization and in Jet you can use it for stream data. In essence, you must implement a StreamSerializer for the objects you emit from your processors and register it in Jet configuration:

Consult the chapter on custom serialization in Hazelcast IMDG’s reference manual for more details.

Note the limitation implied here: the serializers must be registered with Jet on startup because this is how it is supported in Hazelcast IMDG. There is a plan to improve this and allow serializers to be registered on individual Jet jobs.

Jet only serializes an item when it actually sends it to another member. If the target processor is on the same member, it will receive the exact same instance. To catch (de)serialization issues early on, we recommend using a 2-member local Jet cluster for development and testing.

Sharing the same objects between stages is good for performance, but it places a higher degree of responsibility on the developer. You can follow these rules of thumb to avoid concurrency bugs:

Working with immutable objects has its price: you incur allocation and GC pressure by creating the copies and you waste additional CPU time copying the data. This is why we also present fine-grained rules you must follow to stay safe while sharing the data between pipeline stages.

1. Never mutate an item you emitted.

It is never safe to mutate an item you emitted. The downstream stages will see your changes and lose the original data you emitted, at best. Since the changes are concurrent, they may also see completely broken data, if the object is not thread-safe.

One use case where it’s especially tempting to mutate the item after emitting is rolling aggregation. Consider this example where we use mapUsingContext to implement a custom rolling aggregation:

The downstream stage receives the same object each time, but it’s supposed to see it in the state as it was sent. In reality, the aggregating stage keeps changing the object concurrently while the downstream stage tries to observe it. In the best case it will crash with a ConcurrentModificationException, but it may also observe null or partially initialized objects in the list and a number of other bugs.

Jet’s AggregateOperation has a primitive specifically devoted to avoiding this problem: the exportFn. It transforms the accumulator to a new object so the accumulator can be safely updated afterwards. By contrast, finishFn is allowed to return the accumulator itself and Jet calls it only when it knows the whole operation is done and the accumulator won’t be modified again.

There’s another case where a data race can sneak up on you: a many-to-many hash-join, where you enrich an item with several items matching the same join key. Refer to the hash-join section for more details and an example.

2. If you keep using the item you emitted (for reading), don’t mutate it downstream.

There’s little reason to keep reading an item after emitting it, but if you ever have that situation, make sure you don’t mutate it in any downstream stage.

3. If you create a fork in the pipeline (send the output of one stage to several others), no stage after the fork may mutate the data.

Here’s an example with a stream of Person items. We attach two mapping stages to it: one reads the name and the other modifies it. The resulting Jet pipeline will have a data race and behave erratically, sometimes seeing the original name and sometimes the modified one:

By default, Jet runs each internal DAG vertex, roughly equivalent to each step of the computation (such as map or aggregate), at maximum parallelism (equal to the number of CPU cores). This means that even a single Jet job uses quite a lot of parallel tasks. Since Jet’s cooperative tasklets are very cheap to switch between, there’s almost no overhead from this. However, every pair of tasklets that communicate uses a dedicated 1-to-1 concurrent queue so the number of queues scales with the square of the number of CPU cores. The default queue capacity is 1024, which translates to 4-8 kilobytes RAM overhead per tasklet pair and potentially a lot of data items in flight before the queues fill up.

If you experience RAM shortage on the Jet cluster, consider lowering the queue size. This is how you set the default queue size for the whole Jet cluster:

You can also set queue sizes individually on each Core API DAG edge. You must first convert your pipeline to the Core DAG, apply the configuration, and then submit the DAG for execution:

Jet, like Hazelcast IMDG, does not depend on a specific logging framework and has built-in adapters for a variety of logging frameworks. You can also write a new adapter to integrate with loggers Jet doesn’t natively support. To use one of the built-in adapters, set the hazelcast.logging.type property to one of the following:

For example, to configure Jet to use Log4j, you can do one of the following:

or

For more detailed information about how to configure logging, please refer to the IMDG reference manual.

While debugging your pipeline you’ll want to see the output of an individual stage. You can achieve it by using the peek() stage. For example:

If you run it like this:

this is how your output may look:

The logger name of com.hazelcast.jet.impl.processor.PeekWrappedP.filter#1 consists of the following parts:

For more information about logging when using the Core API, see the Best Practices section.

Jet exposes various metrics to facilitate monitoring of the cluster state and of running jobs.

Metrics have associated tags which describe which object the metric applies to. The tags for job metrics typically indicate the specific vertex and processor the metric belongs to.

Each metric instance provided belongs to a particular Jet cluster member, so different cluster members can have their own versions of the same metric with different values.

The metric collection runs in regular intervals on each member, but note that the metric collection on different cluster members happens at different moments in time. So if you try to correlate metrics from different members, they can be from different moments of time.

There are two broad categories of Jet metrics. For clarity we will group them based on significant tags which define their granularity.

Last but not least let’s not forget about the fact that each Jet member is also an IMDG member, so Jet also exposes all the metrics available in IMDG.

Let’s look at these 3 broad categories of metrics in detail.

You can hash-join a stream with up to two enriching streams using the API we demonstrated above. If you have more than two enriching streams, you’ll use the hash-join builder. For example, you may want to enrich a trade with its associated product, broker, and market:

The data type on the hash-joined stage is Tuple2<Trade, ItemsByTag>. The next snippet shows how to use it to access the primary and enriching items:

Hazelcast Jet allows you to enrich your stream directly from a Hazelcast IMap or ReplicatedMap. Since it must look up the data again for each item, performance is lower than with a hash-join, but the data is kept fresh this way. This matters especially for unbounded streaming jobs, where a hash-join would use data frozen in time at the beginning of the job.

If you enrich from a Hazelcast map (IMap or ReplicatedMap) that is stored inside the Jet cluster, you can achieve data locality. For ReplicatedMap that’s trivial because its entire contents are present on every cluster member. IMap, on the other hand, is partitioned so a given member holds only a part of the data. You must give Jet the key-extracting function so it can do the following for each item in your stream:

We didn’t completely avoid the network here, but we replaced the request-response cycle with just a one-way send. We eliminated the cost of the request-response latency and pay just the overhead of network transfer.

In this example we enrich a stream of trades with detailed stock info. The stock info (the enriching dataset) is stored in a Hazelcast IMap so we use groupingKey() to let Jet partition our stream and use local IMap lookups:

In the example above the syntax looks very simple, but it hides a layer of complexity: you can’t just fetch a Hazelcast map instance and add it to the pipeline. It’s a proxy object that’s valid only in the JVM process where you obtained it. mapUsingIMap actually remembers just the name of the map and will obtain a map with the same name from the Jet cluster on which the job runs.

Hazelcast Jet exposes the facility to look up from an external system. In this case you must define a factory object that creates a client instance and through which you will fetch the data from the remote system. Jet will use the factory to create a context object for each Processor in the cluster that executes your transforming step.

The feature is available in two variants:

You can also specify a function that extracts a key from your items using groupingKey(). Even though Jet won’t do any lookup or grouping by that key, it will set up the job so that all the items with the same key get paired with the same context object.

For example, you may be fetching data from a remote Hazelcast cluster. To optimize performance, you can enable the near-cache on the client and you can save a lot of memory by specifying the key function: since Jet executes your enriching code in parallel, each worker has its own Hazelcast client instance, with its own near-cache. With the key function, each near-cache will hold a non-overlapping subset of the keys:

In a similar fashion you can integrate other external systems with a Jet pipeline.

When you use the xUsingY transform to enrich an infinite stream, your output will reflect the fact that the enriching dataset is changing. Two items with the same key may be enriched with different data. This is expected. However, the output may also behave in a surprising way. Say you update the enriching data for some key from A to B. You expect to see a sequence …​A-A-B-B…​ in the output, but when you sort it by timestamp, you can observe any order, such as A-B-A-B or B-B-A-A. This is because Jet doesn’t enforce the processing order to strictly follow the timestamp order of events.

In effect, your lookups will be eventually consistent, but won’t have the monotonic read consistency. Currently there is no feature in Jet that would achieve monotonic reads while enriching an event stream from a changing dataset.

Let’s assume you have these input stages:

You may have expected to find a joining transform in the Pipeline API that outputs a stream of all matching pairs of items:

This would more closely match the semantics of an SQL JOIN, but in the context of a Java API it doesn’t work well. For M page visits joined with N add-to-carts, Jet would have to materialize M * N tuples and feed them into the next stage. It would also be forced to buffer all the data from one stream before receiving the other.

To allow you to get the best performance, Jet strongly couples joining with aggregation and encourages you to frame your solution in these terms.

This doesn’t stop you from getting the "exploded" output with all the joined data, it just makes it a less straightforward option. If your use case can’t be solved any other way than by keeping all individual items, you can specify toList() as the aggregate operation and get all the items in lists:

If you need something else than the full join, you can filter out some pairs of lists. In this example we create a LEFT OUTER JOIN by removing the pairs with empty left-hand list:

If you specifically need to get a stream of all the combinations of matched items, you can add a flatmapping stage:

We used this helper method:

If you need to join more than three streams, you’ll have to use the builder object. For example, your goal may be correlating events coming from different systems, where all the systems serve the same user base. In an online store you may have separate event streams for product page visits, adding-to-cart events, payments, and deliveries. You want to correlate all the events associated with the same user. The example below calculates statistics per category for each user:

The Pipeline API guides you to set up the timestamp policy right after you obtain a source stage. pipeline.drawFrom(someStreamSource) returns a SourceStreamStage which offers just these methods:

Exceptionally, you may need to call withoutTimestamps() on the source stage, then perform some transformations that determine the event timestamps, and then call addTimestamps(timestampFn) to instruct Jet where to find them in the events. Some examples include an enrichment stage that retrieves the timestamps from a side input or flat-mapping the stream to unpack a series of events from each original item. If you do this, however, it will no longer be the source that determines the watermark.

In order to explain windowing, let’s start with a simple batch job, the Word Count. If you have a batch of tweets you want to analyze for word frequencies, this is how the pipeline can look:

To make the same kind of computation work on an infinite stream of tweets, we must add the event timestamps and the specification of the window:

Now we’ve got a Jet job that does live tracking of words currently trending in tweets. The sliding window definition tells Jet to aggregate the events that occurred within the last minute and update this result every second.

Using native timestamps often works well, but it’s not the most precise way of determining the event time. It records the time when the event was submitted to the system that Jet uses as the source and that may happen quite a bit after the event originally occurred. The best approach is to provide Jet with the function that extracts the timestamp directly from the event object. In this case you must also deal with the fact that Jet can receive events in an order different from their timestamp order. We discuss these concerns in the Jet Concepts chapter.

To extract the event timestamp, create the source stage like this:

We specified two things: how to extract the timestamp and how much event "lateness" we want to tolerate. We said that any event we receive can be at most five seconds behind the highest timestamp we already received. If it’s older than that, we’ll have to ignore it. On the flip side, this means that we’ll have to wait for an event that occurred five seconds after the given window’s end before we can assume we have all the data to emit that result.

If you had to allow a lot of event lateness, or if you just use large time windows, you may want to track the progress of a window while it is still accumulating events. You can order Jet to give you, at regular intervals, the current status on all the windows it has some data for, but aren’t yet complete. On our running example of trending words, it may look like this:

In the above snippet we set up an aggregation stage whose window size is one minute and there’s an additional 15-second wait time for the late-coming events. This amounts to waiting up to 75 seconds from receiving a given event to getting the result it contributed to. Therefore we ask Jet to give us updates on the current progress every second.

The output of the windowing stage is in the form of KeyedWindowResult<String, Long>, where String is the word and Long is the frequency of the word in the given window. KeyedWindowResult also has an isEarly property that says whether the result is early or final.

Jet provides some test and development sources which can be used both for batch and streaming jobs.

Jet also provides assertions which can be used for asserting the intermediate or final output of a pipeline. Assertions are implemented as sinks which collect incoming items and then run the desired assertion on the collected items.

Hazelcast Jet can join several streams and simultaneously perform aggregation on all of them. You specify a separate aggregate operation for each input stream and have the opportunity to combine their results when done. You can use aggregate operations provided in the library (see the section on co-aggregation for an example).

If you cannot express your aggregation logic using this approach, you can also specify a custom multi-input aggregate operation that can combine the items into the accumulator immediately as it receives them.

We’ll present a simple example on how to build a custom multi-input aggregate operation. Note that the same logic can also be expressed using separate single-input operations; the point of the example is introducing the API.

Say we are interested in the behavior of users in an online shop application and want to gather the following statistics for each user:

This data is dispersed among separate datasets: PageVisit, AddToCart and Payment. Note that in each case we’re dealing with a simple sum applied to a field in the input item. We can perform a cogroup-and-aggregate transform with the following aggregate operation:

Note how we got an AggregateOperation3 and how it captured each input type. When we use it as an argument to a cogroup-and-aggregate transform, the compiler will ensure that the ComputeStage s we attach it to have the correct type and are in the correct order.

On the other hand, if you use the co-aggregation builder object, you’ll construct the aggregate operation by calling andAccumulate(tag, accFn) with all the tags you got from the co-aggregation builder, and the static type will be just AggregateOperation. The compiler won’t be able to match up the inputs to their treatment in the aggregate operation.

The first decision when building a Jet computation job is whether it will deal with bounded or unbounded data. A typical example of a bounded resource is a persistent storage system, whereas an unbounded one is usually like a FIFO queue, discarding old data. This is true both for sources and sinks.

Bounded data is handled in batch jobs and there are less concerns to deal with. Examples of finite resources are the Hazelcast IMap/ICache and the Hadoop Distributed File System (HDFS). In the unbounded category the most popular choice is Kafka, but a Hazelcast IMap/ICache can also be used as an infinite source of update events (via the Event Journal feature). You can also set up an IMap/ICache as a sink for an infinite amount of data, either by ensuring that the size of the keyset will be finite or by allowing the eviction of old entries.

Most finite data sources are replayable because they come from persistent storage. You can easily replay the whole dataset. However, an infinite data source may be of such nature that it can be consumed only once. An example is the TCP socket connector. Such sources are bad at fault tolerance: if anything goes wrong during the computation, it cannot be retried.

A distributed computation engine prefers to work with distributed data resources. If the resource is not distributed, all Jet members will have to contend for access to a single endpoint. Kafka, HDFS, IMap and ICache are all distributed. On the other hand, an IList is not: it resides on a single member. When used as a source, only one Jet member pulls its data. When used as a sink, all Jet members send their data to the one that holds it.

A file source/sink operating in local mode is a sort of a "manually distributed" resource, each member accessing its own local filesystem. You have to manually arrange the files so that on each member there is a subset of the full dataset. When used as a sink, you have to manually gather all the pieces that Jet created.

The file source/sink can also operate in shared mode, accessing a shared filesystem mounted as a local directory.

If you’re looking to achieve record-breaking throughput for your application, you’ll have to think carefully how close you can deliver your data to the location where Jet will consume and process it. For example, if your source is HDFS, you should align the topologies of the Hadoop and Jet clusters so that each machine that hosts an HDFS member also hosts a Jet member. Jet will automatically figure this out and arrange for each member to consume only the slice of data stored locally.

If you’re using IMap/ICache as data sources, you have two basic choices: have Jet connect to a Hazelcast IMDG cluster, or use Jet itself to host the data (since a Jet cluster is at the same time a Hazelcast IMDG cluster). In the second case Jet will automatically ensure a data-local access pattern, but there’s a caveat: if the Jet job causes an error of unrestricted scope, such as OutOfMemoryError or StackOverflowError, it will have unpredictable consequences for the state of the whole Jet member, jeopardizing the integrity of the data stored on it.

To access a Hazelcast IMDG cluster separate from the Jet cluster, you have to provide Hazelcast client configuration for the connection. In this simple example we use programmatic configuration to draw from and drain to remote IMap and ICache. Just for variety, we funnel the data from IMap to ICache and vice versa:

For a full discussion on how to configure your client connection, refer to the Hazelcast IMDG documentation on this topic.

If your use case calls for some filtering and/or transformation of the data you retrieve, you can optimize the traffic volume by providing a filtering predicate and an arbitrary transformation function to the source connector itself and they’ll get applied on the remote side, before sending:

The same optimization works on a local IMap, too, but has less impact. However, Hazelcast IMDG goes a step further in optimizing your filtering and mapping to a degree that matters even locally. If you don’t need fully general functions, but can express your predicate via Predicates or PredicateBuilder, they will create a specialized predicate instance that can test the object without deserializing it. Similarly, if the mapping you need is of a constrained kind where you just extract one or more object fields (attributes), you can specify a projection instead of a general mapping lambda: Projections.singleAttribute() or Projections.multiAttribute(). These will extract the listed attributes without deserializing the whole object. For these optimizations to work, however, your objects must employ Hazelcast’s portable serialization. They are especially relevant if the volume of data you need in the Jet job is significantly less than the volume of the stored data.

Note that the above feature is not available on ICache. It is, however, available on `ICache’s event journal, which we introduce next.

You can use IMap/ICache as sources of infinite event streams. For this to work you have to enable the Event Journal on your data structure. This is a feature you set in the Jet/IMDG instance configuration, which means you cannot change it while the cluster is running.

This is how you enable the Event Journal on an IMap:

The default journal capacity is 10,000 and the default time-to-live is 0 (which means “unlimited”). Since the entire event journal is kept in RAM, you should take care to adjust these values to match your use case.

The configuration API for ICache is identical:

Once properly configured, you use Event Journal sources like this:

IMap and ICache are on an equal footing here. The second argument, START_FROM_CURRENT here, means "start receiving from events that occur after the processing starts". If you specify START_FROM_OLDEST, you’ll get all the events still on record.

This version of methods will only emit ADDED and UPDATED event types. Also, it will map the event object to simple Map.Entry with the key and new value. To make a different choice, use the overloads that allow you to specify your own projection and filtering. For example, you can request all event types and full event objects:

Note the type of the stream element: EventJournalMapEvent and EventJournalCacheEvent. These are almost the same and have these methods:

The only difference is the return type of getType() which is specific to each kind of structure and gives detailed insight into what kind of event it reports. Add, remove and update are the basic ones, but there are also evict, clear, expire and some others.

Finally, you can get all of the above from a map/cache in another cluster, you just have to prepend remote to the source names and add a ClientConfig, for example:

When you use an IMap as a sink, instead of just pushing the data into it you may have to merge the new with the existing data or delete the existing data. Hazelcast Jet supports this with map-updating sinks which rely on Hazelcast IMDG’s Entry Processor feature. An entry processor allows you to atomically execute a piece of code against a map entry, in a data-local manner.

The updating sinks come in three variants:

The Kafka source emits entries of type Map.Entry<Key,Value> which can be transformed using an optional mapping function. It never completes. The job will end only if explicitly cancelled or aborted due to an error.

Internally Jet creates one KafkaConsumer per Processor instance using the supplied properties. Jet uses manual partition assignment to arrange the available Kafka partitions among the available processors and will ignore the group.id property.

Currently there is a requirement that the global parallelism of the Kafka source be at most the number of partitions you are subscribing to. The local parallelism of the Kafka source is 2 and if your Jet cluster has 4 members, this means that a minimum of 8 Kafka partitions must be available.

If any new partitions are added while the job is running, Jet will automatically assign them to the existing processors and consume them from the beginning.

The Kafka source supports snapshots. Upon each snapshot it saves the current offset for each partition. When the job is restarted from a snapshot, the source can continue reading from the saved offset.

If snapshots are disabled, the source will commit the offset of the last record it read to the Kafka cluster. Since the fact that the source read an item doesn’t mean that the whole Jet pipeline processed it, this doesn’t guarantee against data loss.

The Kafka sink creates one KafkaProducer per cluster member and shares it among all the sink processors on that member. You can provide a mapping function that transforms the items the sink receives into `ProducerRecord`s.

Apache Kafka introduced client backward compatibility with version 1.0.0. The compatibility is two way, new brokers support older clients and new clients support older broker.

The Kafka sink and source are based on version 2.2.0, this means Kafka connector will work with any client and broker having version equal to or greater than 1.0.0.

Jet will split the input data across the cluster, with each processor instance reading a part of the input. If the Jet nodes are running along the HDFS datanodes, then Jet can make use of data locality by reading the blocks locally where possible. This can bring a significant increase in read speed.

Each processor will write to a different file in the output folder identified by the unique processor id. The files will be in a temporary state until the job is completed and will be committed when the job is complete. For streaming jobs, they will be committed when the job is cancelled. We have plans to introduce a rolling sink for HDFS in the future to have better streaming support.

Hadoop types implement their own serialization mechanism through the use of Writable. Jet provides an adapter to register a Writable for Hazelcast serialization without having to write additional serialization code. To use this adapter, you can register your own Writable types by extending WritableSerializerHook and registering the hook.

When submitting JARs along with a Job, sending Hadoop JARs should be avoided and instead Hadoop JARs should be present on the classpath of the running members. Hadoop JARs contain some JVM hooks and can keep lingering references inside the JVM long after the job has ended, causing memory leaks.

The file sources are designed to work with the local and shared file systems. For local file system, the sources expect to see on each member just the files that member should read. You can achieve the effect of a distributed source if you manually prepare a different set of files on each member. For shared file system, the sources split the work so that each member will read a part of the files.

There are two flavors of the file source: bounded and unbounded.

Here’s an example with the bounded source:

This will log on each Jet member the contents of all the files in the specified directory. When the source reads all the files, the job completes. If the files change while the job is running, the behavior is undefined.

Here’s an example with the unbounded source:

It will watch the directory for changes. It will emit only new contents added after startup: both new files and new content appended to existing ones. Files must be updated in an append-only fashion; if the existing content changes, the behavior is undefined.

If you delete the watched directory, the job will complete.

The file sink can work with either a local or a shared network file system. Each member will write to different file names. You can achieve the effect of a distributed sink if you manually collect all the output files on all members and combine their contents.

Here’s a small example of usage:

Hazelcast Jet provides Apache Avro file source and sink for batch processing jobs. This is a specialized version of file connector. Avro connector assumes the data is in the Avro Object Container File format.

The socket source opens a blocking client TCP/IP socket and receives data over it. The data must be lines of plain text.

Each underlying worker of the Socket Source connector opens its own client socket and asks for data from it. The user supplies the host:port connection details. The server side should ensure a meaningful dispersion of data among all the connected clients, but how it does it is outside of Jet’s control.

Here’s a simple example:

You can study a comprehensive code sample including a sample socket server using Netty.

The JDBC source comes in two versions:

The JDBC sink connects to the specified database using the given connection supplier, prepares a statement using the given update query and inserts/updates the items.

The update query should be a parametrized query. The bind function will receive a PreparedStatement created for this query and should bind parameters to it. It should not execute the query, call commit or any other method.

The records will be committed after each batch of records and a batch mode will be used (if the driver supports it). Auto-commit will be disabled on the connection.

In case of an SQLException the processor will automatically try to reconnect and the job won’t fail, except for the SQLNonTransientException subclass. The default local parallelism for this sink is 1.

No state is saved to snapshot for this sink. After the job is restarted, the items will likely be duplicated, providing an at-least-once guarantee. For this reason you should not use INSERT statement which can fail on duplicate primary key. Rather use an insert-or-update statement that can tolerate duplicate writes.

The following code snippet shows writing the Person objects to a database table:

The JMS source opens a connection to the JMS server for each member. Then each underlying worker of the source creates a session and a message consumer using that connection. The user supplies necessary functions to create the connection, session and message consumer.

The JMS source uses non-blocking API to receive the messages and transforms each message to a desired output object using supplied projection function. Source flushes the session after receiving each message using the supplied flush function.

The following code snippets show streaming messages from a JMS queue and a JMS topic using ActiveMQ JMS Client.

The JMS topic is a non-distributed source, if messages are consumed by multiple consumer, all of them will get the same messages. Therefore the source operates on a single member with local parallelism of 1. Setting local parallelism to a value other than 1 causes IllegalArgumentException.

The JMS sink opens a connection to the JMS server for each member. Then each underlying worker of the sink creates a session and a message producer using that connection. The user supplies necessary functions and parameters to create the connection, session and message producer.

The JMS sink uses the supplied function to create a Message object for each input item and sends this message using the supplied send function. After a batch of messages is sent, sink flushes the session using the supplied flush function.

The following code snippets show writing to a JMS queue and a JMS topic using ActiveMQ JMS Client.

S3 source distributes the workload by splitting the objects to each processor instance. The source reads each object line by line and emits desired output to downstream using provided transform function.

Below example will log on each Jet member the contents of all the objects in the specified bucket. When the source reads all the files, the job completes. If the objects change while the job is running, the behavior is undefined.

S3 sink writes items to the specified bucket by transforming each item to a line using given transform function. The sink creates an object in the bucket for each processor instance. Name of the file will include an user provided prefix (if defined) and followed by the processor’s global index, for example the processor having the index 2 with prefix my-object- will create the object my-object-2.

Here’s a small example of usage:

AWS S3 can be registered as a filesystem to HDFS which then can be used as a source or sink via our hadoop module. See Hadoop-AWS module for more information on the dependencies and configuration.

Below example will read from a bucket, transform each line to uppercase and write to another bucket using the hadoop module.

To make your custom source connector you need two basic ingredients:

Jet repeatedly calls fillBufferFn whenever it needs more data items. Optimally, the function will fill the buffer with the items it can acquire without blocking. A hundred items at a time is enough to eliminate any per-call overheads within Jet. The function may block as well (it’s running inside a non-cooperative processor), but taking longer than a second to complete can have negative effects on the overall performance of the processing pipeline.

To make your custom sink connector you need two basic ingredients:

This is a simple example with a file sink:

This will create the file output.txt on each member, so the overall job output consists of the contents of all these files put together.

Note that we’re using blocking IO here. Jet will use non-cooperative processors for the sink, so we’re allowed to do that. For good overall job performance we should still ensure that the call receiveFn doesn’t take more than a second to complete.

In the above example our state object is a PrintWriter, which has internal buffering. Jet allows us to make buffering a first-class concern and deal with it explicitly by taking an optional flushFn which it will call at regular intervals. Here’s our example modified to do explicit buffering and flushing:

In this case we don’t need the destroyFn because we keep the file closed while not appending to it. Through the use of buffering we drown out the overhead of opening and closing the file each time and we get an overall more robust solution.

Programmatic configuration is the simplest way to configure Jet. You instantiate a JetConfig and set the desired properties. For example, the following will configure Jet to use only two threads in its cooperative multithreading pool:

If you want to specify your own configuration file to create JetConfig, Hazelcast Jet supports several ways including filesystem, classpath, InputStream and in-memory string.

Building JetConfig from the XML declarative configuration:

Building JetConfig from the YAML declarative JetConfiguration:

If you don’t pass an explicit JetConfig object when constructing a Jet instance, it will perform the following lookup procedure:

An example hazelcast-jet.xml is shown below :

The identical part of the configuration extracted from hazelcast-jet.yaml is shown as below:

In your Hazelcast Jet declarative configuration, you can use variables to set the values of the elements. This is valid when you set a system property programmatically or use the command line interface. You can use a variable in the declarative configuration to access the values of the system properties you set.

For example, see the following command that sets two system properties.

Let’s get the values of these system properties in the declarative configuration of Hazelcast Jet, as shown below.

In the XML configuration:

In the YAML configuration:

If you do not want to rely on the system properties, you can use the XmlJetConfigBuilder or YamlJetConfigBuilder and explicitly set a Properties instance, as shown below.

Hazelcast Jet supports variable replacers which are used to replace custom strings during loading the configuration.

Hazelcast Jet and Hazelcast IMDG share the same syntax for variable replacers. You can configure your variable replacers in the Hazelcast Jet configuration via XML or YAML. Please see the Variable Replacers section in the Hazelcast IMDG Reference Manual for more information and examples on the variable replacers.

You can configure the underlying Hazelcast IMDG member as follows:

Similarly, you can also configure the underlying Hazelcast client:

Hazelcast IMDG can also be configured declaratively. Please refer to the Hazelcast IMDG Reference Manual for information on how to do this.

You can declare Hazelcast Jet objects using the default Spring beans namespace. Here is an example for a Hazelcast Jet Instance declaration:

Hazelcast Jet provides its own Spring config schema. Add the namespace declaration xmlns:jet=“http://www.hazelcast.com/schema/jet-spring” to the beans element and then use the jet namespace prefix. Make sure you added hazelcast-jet-spring.jar to the classpath.

Here’s how your namespace and schema instance declarations may look: