Forem: Hazelcast

Hazelcast + Kibana: best buddies for exploring and visualizing data

Nicolas Fränkel — Thu, 28 Oct 2021 06:39:14 +0000

A lot, if not all, of data science projects, require some data visualization front-end to display the results for humans to analyze. Python seems to boast the most potent libraries, but do not lose hope if you're a Java developer (or if you're proficient in another language as well). In this post, I will describe how you can benefit from such a data visualization front-end without writing a single line of code.

The use case: changes from Wikipedia

I infer that you are already familiar with Wikipedia. If you do not, Wikipedia is an online encyclopedia curated by the community. In their own words:

>Wikipedia is a free content, multilingual online encyclopedia written and maintained by a community of volunteer contributors through a model of open collaboration, using a wiki-based editing system.

The above is actually an excerpt of the Wikipedia entry on Wikipedia itself. Very meta.

The idea is to let anybody write anything on any subject and let the community decide whether the piece improves the body of knowledge - or not. You can think about this system as a worldwide Git review.

Even with this in place, it would be easy to overflow the reviewing capacity of the community by sending lots and lots of changes. To prevent such abuse, would-be contributors need to create an account first. However, it adds a layer of friction. If I want to contribute by fixing a typo, adding an image, or any other tiny task, creating my account would be more time-consuming than contributing. To allow for one-time contributions, Wikipedia allows anonymous changes. However, we get back to square one regarding abuses. To cover that, Wikipedia logs your IP in that case. The IP will appear in the change history instead of the account's name.

Now, my use-case is about visualizing worldwide, anonymous contributions. I'll first read the data from Wikipedia, filter out changes by authenticated accounts, infer the location of the change, infer the language of the change, and then display them on a worldwide map. From this point, I'd explore the changes visually and say that language and location match somehow.

We are going to achieve that by following a step-by-step process.

Waiting On the World to Change

The first step on our hands is actually to get data in, i.e., to get the changes from Wikipedia into our data store. It's pretty straightforward, as Wikipedia itself provides its changes on a dedicated Recent Changes page. If you press the "Live Update" button, you can see the list is updated in real-time (or very close to). Here's a screenshot of the changes at the time of the writing of this post:

Now is the time to create a data pipeline to get this data in Hazelcast. Note that if you want to follow along, the project is readily available on GitHub.

Wikipedia provides changes through Server-Sent Events. In short, with SSE, you register a client to the endpoint, and every time new data comes in, you are notified and can act accordingly. On the JVM, a couple of SSE-compatible clients are available, including Spring WebClient. Instead, I chose to use OkHttp EventSource because it's lightweight - it only depends on OkHttp, and its usage is relatively straightforward.

Here's an excerpt from the POM:

<dependency>
    <groupId>com.hazelcast</groupId>
    <artifactId>hazelcast</artifactId>
    <version>${hazelcast.version}</version>
</dependency>
<dependency>
    <groupId>com.launchdarkly</groupId>
    <artifactId>okhttp-eventsource</artifactId>
    <version>2.3.2</version>
</dependency>

Getting data in

Hazelcast data pipelines work by regularly polling the source. With an HTTP endpoint, that's straightforward, but with SSE, not so much as SSE relies on subscription. Hence, we need to implement a custom Source and design it around an internal queue to store the changes as they arrive, while polling will dequeue and send them further down the pipeline.

We design the code around the following components:

Context manages the subscription. It creates a new WikipediaChangeEventHandler instance and registers it as an observer of the SSE stream.
WikipediaChangeEventHandler is the subscribing part. Every time a change happens, it gets notified and queues the change payload in its internal queue.
The Hazelcast engine calls Call at regular intervals. When it happens, it dequeues items from WikipediaChangeEventHandler, transforms the plain string into a JSONObject, and puts the latter in the data pipeline buffer.

From a dynamic point of view, the system can be modeled as:

Running the code outputs something like this:

{"server_script_path":"/w","server_name":"en.wikipedia.org","$schema":"/mediawiki/recentchange/1.0.0","bot":false,"wiki":"enwiki","type":"categorize","title":"Category:Biography articles without listas parameter","meta":{"dt":"2021-07-28T04:07:40Z","partition":0,"offset":363427323,"stream":"mediawiki.recentchange","domain":"en.wikipedia.org","topic":"codfw.mediawiki.recentchange","id":"01592c7a-03f1-46cd-9472-3bbe63aff0ec","uri":"https://en.wikipedia.org/wiki/Category:Biography_articles_without_listas_parameter","request_id":"b49c3b98-2064-44da-aab4-ab7b3bf65bdd"},"namespace":14,"comment":"[[:Talk:Jeff S. Klotz]] removed from category","id":1406951122,"server_url":"https://en.wikipedia.org","user":"Lepricavark","parsedcomment":"<a href=\"/wiki/Talk:Jeff_S._Klotz\" title=\"Talk:Jeff S. Klotz\">Talk:Jeff S. Klotz<\/a> removed from category","timestamp":1627445260}
{"server_script_path":"/w","server_name":"commons.wikimedia.org","$schema":"/mediawiki/recentchange/1.0.0","bot":true,"wiki":"commonswiki","type":"categorize","title":"Category:Flickr images reviewed by FlickreviewR 2","meta":{"dt":"2021-07-28T04:07:42Z","partition":0,"offset":363427324,"stream":"mediawiki.recentchange","domain":"commons.wikimedia.org","topic":"codfw.mediawiki.recentchange","id":"68f3a372-112d-4dae-af8f-25d88984f1d8","uri":"https://commons.wikimedia.org/wiki/Category:Flickr_images_reviewed_by_FlickreviewR_2","request_id":"1a132610-85e0-4954-9329-9e44691970aa"},"namespace":14,"comment":"[[:File:Red squirrel (51205279267).jpg]] added to category","id":1729953358,"server_url":"https://commons.wikimedia.org","user":"FlickreviewR 2","parsedcomment":"<a href=\"/wiki/File:Red_squirrel_(51205279267).jpg\" title=\"File:Red squirrel (51205279267).jpg\">File:Red squirrel (51205279267).jpg<\/a> added to category","timestamp":1627445262}
{"server_script_path":"/w","server_name":"commons.wikimedia.org","$schema":"/mediawiki/recentchange/1.0.0","bot":true,"wiki":"commonswiki","type":"categorize","title":"Category:Flickr review needed","meta":{"dt":"2021-07-28T04:07:42Z","partition":0,"offset":363427325,"stream":"mediawiki.recentchange","domain":"commons.wikimedia.org","topic":"codfw.mediawiki.recentchange","id":"b4563ed9-a6f2-40de-9e71-c053f5352846","uri":"https://commons.wikimedia.org/wiki/Category:Flickr_review_needed","request_id":"1a132610-85e0-4954-9329-9e44691970aa"},"namespace":14,"comment":"[[:File:Red squirrel (51205279267).jpg]] removed from category","id":1729953359,"server_url":"https://commons.wikimedia.org","user":"FlickreviewR 2","parsedcomment":"<a href=\"/wiki/File:Red_squirrel_(51205279267).jpg\" title=\"File:Red squirrel (51205279267).jpg\">File:Red squirrel (51205279267).jpg<\/a> removed from category","timestamp":1627445262}
{"server_script_path":"/w","server_name":"www.wikidata.org","$schema":"/mediawiki/recentchange/1.0.0","minor":false,"bot":true,"wiki":"wikidatawiki","length":{"new":31968,"old":31909},"type":"edit","title":"Q40652","revision":{"new":1468164253,"old":1446892882},"patrolled":true,"meta":{"dt":"2021-07-28T04:07:43Z","partition":0,"offset":363427326,"stream":"mediawiki.recentchange","domain":"www.wikidata.org","topic":"codfw.mediawiki.recentchange","id":"70784dde-0360-4292-9f62-81323ced9aa7","uri":"https://www.wikidata.org/wiki/Q40652","request_id":"f9686303-ffed-4c62-8532-bf870288ff55"},"namespace":0,"comment":"/* wbsetaliases-add:1|zh */ 蒂托, [[User:Cewbot#Import labels/aliases|import label/alias]] from [[zh:巴西國家足球隊]], [[zh:何塞·保罗·贝塞拉·马希尔·儒尼奥尔]], [[zh:2018年國際足協世界盃參賽球員名單]], [[zh:埃德爾·米利唐]], [[zh:加布里埃爾·馬丁內利]], [[zh:2019年南美超级德比杯]], [[zh:2019年美洲杯决赛]], [[zh:2019年美洲杯参赛名单]], [[zh:2021年美洲杯B组]], [[zh:2021年美洲國家盃決賽]]","id":1514670479,"server_url":"https://www.wikidata.org","user":"Cewbot","parsedcomment":"\u200e<span dir=\"auto\"><span class=\"autocomment\">Added Chinese alias: <\/span><\/span> 蒂托, <a href=\"/wiki/User:Cewbot#Import_labels/aliases\" title=\"User:Cewbot\">import label/alias<\/a> from <a href=\"https://zh.wikipedia.org/wiki/%E5%B7%B4%E8%A5%BF%E5%9C%8B%E5%AE%B6%E8%B6%B3%E7%90%83%E9%9A%8A\" class=\"extiw\" title=\"zh:巴西國家足球隊\">zh:巴西國家足球隊<\/a>, <a href=\"https://zh.wikipedia.org/wiki/%E4%BD%95%E5%A1%9E%C2%B7%E4%BF%9D%E7%BD%97%C2%B7%E8%B4%9D%E5%A1%9E%E6%8B%89%C2%B7%E9%A9%AC%E5%B8%8C%E5%B0%94%C2%B7%E5%84%92%E5%B0%BC%E5%A5%A5%E5%B0%94\" class=\"extiw\" title=\"zh:何塞·保罗·贝塞拉·马希尔·儒尼奥尔\">zh:何塞·保罗·贝塞拉·马希尔·儒尼奥尔<\/a>, <a href=\"https://zh.wikipedia.org/wiki/2018%E5%B9%B4%E5%9C%8B%E9%9A%9B%E8%B6%B3%E5%8D%94%E4%B8%96%E7%95%8C%E7%9B%83%E5%8F%83%E8%B3%BD%E7%90%83%E5%93%A1%E5%90%8D%E5%96%AE\" class=\"extiw\" title=\"zh:2018年國際足協世界盃參賽球員名單\">zh:2018年國際足協世界盃參賽球員名單<\/a>, <a href=\"https://zh.wikipedia.org/wiki/%E5%9F%83%E5%BE%B7%E7%88%BE%C2%B7%E7%B1%B3%E5%88%A9%E5%94%90\" class=\"extiw\" title=\"zh:埃德爾·米利唐\">zh:埃德爾·米利唐<\/a>, <a href=\"https://zh.wikipedia.org/wiki/%E5%8A%A0%E5%B8%83%E9%87%8C%E5%9F%83%E7%88%BE%C2%B7%E9%A6%AC%E4%B8%81%E5%85%A7%E5%88%A9\" class=\"extiw\" title=\"zh:加布里埃爾·馬丁內利\">zh:加布里埃爾·馬丁內利<\/a>, <a href=\"https://zh.wikipedia.org/wiki/2019%E5%B9%B4%E5%8D%97%E7%BE%8E%E8%B6%85%E7%BA%A7%E5%BE%B7%E6%AF%94%E6%9D%AF\" class=\"extiw\" title=\"zh:2019年南美超级德比杯\">zh:2019年南美超级德比杯<\/a>, <a href=\"https://zh.wikipedia.org/wiki/2019%E5%B9%B4%E7%BE%8E%E6%B4%B2%E6%9D%AF%E5%86%B3%E8%B5%9B\" class=\"extiw\" title=\"zh:2019年美洲杯决赛\">zh:2019年美洲杯决赛<\/a>, <a href=\"https://zh.wikipedia.org/wiki/2019%E5%B9%B4%E7%BE%8E%E6%B4%B2%E6%9D%AF%E5%8F%82%E8%B5%9B%E5%90%8D%E5%8D%95\" class=\"extiw\" title=\"zh:2019年美洲杯参赛名单\">zh:2019年美洲杯参赛名单<\/a>, <a href=\"https://zh.wikipedia.org/wiki/2021%E5%B9%B4%E7%BE%8E%E6%B4%B2%E6%9D%AFB%E7%BB%84\" class=\"extiw\" title=\"zh:2021年美洲杯B组\">zh:2021年美洲杯B组<\/a>, <a href=\"https://zh.wikipedia.org/wiki/2021%E5%B9%B4%E7%BE%8E%E6%B4%B2%E5%9C%8B%E5%AE%B6%E7%9B%83%E6%B1%BA%E8%B3%BD\" class=\"extiw\" title=\"zh:2021年美洲國家盃決賽\">zh:2021年美洲國家盃決賽<\/a>","timestamp":1627445263}
{"server_script_path":"/w","server_name":"www.wikidata.org","$schema":"/mediawiki/recentchange/1.0.0","minor":false,"bot":true,"wiki":"wikidatawiki","length":{"new":239,"old":161},"type":"edit","title":"Q107674623","revision":{"new":1468164250,"old":1468164243},"patrolled":true,"meta":{"dt":"2021-07-28T04:07:43Z","partition":0,"offset":363427327,"stream":"mediawiki.recentchange","domain":"www.wikidata.org","topic":"codfw.mediawiki.recentchange","id":"40260137-ee52-4a67-b024-22d3cf86907a","uri":"https://www.wikidata.org/wiki/Q107674623","request_id":"db6e073a-19f6-4658-9425-7992b34b4208"},"namespace":0,"comment":"/* wbsetlabel-add:1|de */ Favolaschia filopes","id":1514670480,"server_url":"https://www.wikidata.org","user":"SuccuBot","parsedcomment":"\u200e<span dir=\"auto\"><span class=\"autocomment\">Bezeichnung für [de] hinzugefügt: <\/span><\/span> Favolaschia filopes","timestamp":1627445263}
{"server_script_path":"/w","server_name":"ko.wikipedia.org","$schema":"/mediawiki/recentchange/1.0.0","minor":true,"bot":true,"wiki":"kowiki","length":{"new":1158,"old":1161},"type":"edit","title":"이시다테 야스키","revision":{"new":29895993,"old":26098259},"meta":{"dt":"2021-07-28T04:07:43Z","partition":0,"offset":363427328,"stream":"mediawiki.recentchange","domain":"ko.wikipedia.org","topic":"codfw.mediawiki.recentchange","id":"c23bdb77-e88c-48d3-9d24-3c4dd8ef1dbf","uri":"https://ko.wikipedia.org/wiki/%EC%9D%B4%EC%8B%9C%EB%8B%A4%ED%85%8C_%EC%95%BC%EC%8A%A4%ED%82%A4","request_id":"0010e77b-fbcd-4de8-a5ad-4616adbbd6d4"},"namespace":0,"comment":"봇: 분류 이름 변경 (분류:1984년 태어남 → [[분류:1984년 출생]])","id":56333828,"server_url":"https://ko.wikipedia.org","user":"TedBot","parsedcomment":"봇: 분류 이름 변경 (분류:1984년 태어남 → <a href=\"/wiki/%EB%B6%84%EB%A5%98:1984%EB%85%84_%EC%B6%9C%EC%83%9D\" title=\"분류:1984년 출생\">분류:1984년 출생<\/a>)","timestamp":1627445263}
{"server_script_path":"/w","server_name":"commons.wikimedia.org","$schema":"/mediawiki/recentchange/1.0.0","minor":false,"bot":true,"wiki":"commonswiki","length":{"new":3864,"old":527},"type":"edit","title":"File:Albizia kalkora 06.jpg","revision":{"new":577195372,"old":577193453},"patrolled":true,"meta":{"dt":"2021-07-28T04:07:44Z","partition":0,"offset":363427329,"stream":"mediawiki.recentchange","domain":"commons.wikimedia.org","topic":"codfw.mediawiki.recentchange","id":"1a7fcb55-dec7-4303-b757-19f6a6a4dcdd","uri":"https://commons.wikimedia.org/wiki/File:Albizia_kalkora_06.jpg","request_id":"7f841b4a-ac70-4c2b-a148-bc07696ccf7a"},"namespace":6,"comment":"/* wbeditentity-update:0| */ Adding structured data: date, camera, author, copyright & source","id":1729953360,"server_url":"https://commons.wikimedia.org","user":"BotMultichillT","parsedcomment":"\u200e<span dir=\"auto\"><span class=\"autocomment\">Changed an entity: <\/span><\/span> Adding structured data: date, camera, author, copyright & source","timestamp":1627445264}
{"server_script_path":"/w","server_name":"id.wikipedia.org","$schema":"/mediawiki/recentchange/1.0.0","minor":true,"bot":true,"wiki":"idwiki","length":{"new":977,"old":962},"type":"edit","title":"Euporus linearis","revision":{"new":18801346,"old":16068468},"patrolled":true,"meta":{"dt":"2021-07-28T04:07:43Z","partition":0,"offset":363427330,"stream":"mediawiki.recentchange","domain":"id.wikipedia.org","topic":"codfw.mediawiki.recentchange","id":"6c3882f9-9fd0-4f43-ab69-e538762c7981","uri":"https://id.wikipedia.org/wiki/Euporus_linearis","request_id":"dea59b42-7c97-4cbc-9384-5d8836a981ec"},"namespace":0,"comment":"[[Wikipedia:Bot|Bot]]: fixed → [[Kategori:Taxonbar tanpa parameter from|taxonbar tanpa parameter from]]","id":42309169,"server_url":"https://id.wikipedia.org","user":"HsfBot","parsedcomment":"<a href=\"/wiki/Wikipedia:Bot\" title=\"Wikipedia:Bot\">Bot<\/a>: fixed → <a href=\"/wiki/Kategori:Taxonbar_tanpa_parameter_from\" title=\"Kategori:Taxonbar tanpa parameter from\">taxonbar tanpa parameter from<\/a>","timestamp":1627445263}
{"server_script_path":"/w","server_name":"www.wikidata.org","$schema":"/mediawiki/recentchange/1.0.0","minor":false,"bot":false,"wiki":"wikidatawiki","length":{"new":25025,"old":24908},"type":"edit","title":"Q80075231","revision":{"new":1468164255,"old":1467697536},"patrolled":true,"meta":{"dt":"2021-07-28T04:07:44Z","partition":0,"offset":363427331,"stream":"mediawiki.recentchange","domain":"www.wikidata.org","topic":"codfw.mediawiki.recentchange","id":"720f6507-1ea1-4665-b1b9-1665c97450a9","uri":"https://www.wikidata.org/wiki/Q80075231","request_id":"43b7d511-007f-4005-a562-5002c7e0aff4"},"namespace":0,"comment":"/* wbsetdescription-add:1|dv */ އަކުއިލާ ނަކަތުގައިވާ ތަރިއެއް, [[:toollabs:quickstatements/#/batch/60416|batch #60416]]","id":1514670481,"server_url":"https://www.wikidata.org","user":"EN-Jungwon","parsedcomment":"\u200e<span dir=\"auto\"><span class=\"autocomment\">Added [dv] description: <\/span><\/span> އަކުއިލާ ނަކަތުގައިވާ ތަރިއެއް, <a href=\"https://iw.toolforge.org/quickstatements/#.2Fbatch.2F60416\" class=\"extiw\" title=\"toollabs:quickstatements/\">batch #60416<\/a>","timestamp":1627445264}
{"server_script_path":"/w","server_name":"www.wikidata.org","$schema":"/mediawiki/recentchange/1.0.0","minor":false,"bot":false,"wiki":"wikidatawiki","length":{"new":5312,"old":4884},"type":"edit","title":"Q85766437","revision":{"new":1468164246,"old":1342535335},"patrolled":true,"meta":{"dt":"2021-07-28T04:07:42Z","partition":0,"offset":363427332,"stream":"mediawiki.recentchange","domain":"www.wikidata.org","topic":"codfw.mediawiki.recentchange","id":"ad173600-09b7-4ccd-9490-4a60f6a432ea","uri":"https://www.wikidata.org/wiki/Q85766437","request_id":"1228a17e-2baa-46cc-a3bc-2049a62982c9"},"namespace":0,"comment":"/* wbcreateclaim-create:1| */ [[Property:P7937]]: [[Q7366]], [[:toollabs:quickstatements/#/batch/60404|batch #60404]]","id":1514670483,"server_url":"https://www.wikidata.org","user":"Moebeus","parsedcomment":"\u200e<span dir=\"auto\"><span class=\"autocomment\">Created claim: <\/span><\/span> <a href=\"/wiki/Property:P7937\" title=\"Property:P7937\">Property:P7937<\/a>: <a href=\"/wiki/Q7366\" title=\"Q7366\">Q7366<\/a>, <a href=\"https://iw.toolforge.org/quickstatements/#.2Fbatch.2F60404\" class=\"extiw\" title=\"toollabs:quickstatements/\">batch #60404<\/a>","timestamp":1627445262}
{"server_script_path":"/w","server_name":"www.wikidata.org","$schema":"/mediawiki/recentchange/1.0.0","minor":false,"bot":false,"wiki":"wikidatawiki","length":{"new":5134,"old":5126},"type":"edit","title":"Q12444793","revision":{"new":1468164254,"old":1413396080},"patrolled":false,"meta":{"dt":"2021-07-28T04:07:43Z","partition":0,"offset":363427333,"stream":"mediawiki.recentchange","domain":"www.wikidata.org","topic":"codfw.mediawiki.recentchange","id":"c01d52c5-c476-4554-814d-513342e04686","uri":"https://www.wikidata.org/wiki/Q12444793","request_id":"6d0a32b9-1234-4c8e-a02a-d92608f06d33"},"namespace":0,"comment":"/* wbsetdescription-set:1|hi */ भारत के उत्तराखण्ड राज्य का एक गाँव bikash","id":1514670482,"server_url":"https://www.wikidata.org","user":"2409:4061:219C:613E:DFD9:6BD4:F234:E7E0","parsedcomment":"\u200e<span dir=\"auto\"><span class=\"autocomment\">बदला [hi] विवरण: <\/span><\/span> भारत के उत्तराखण्ड राज्य का एक गाँव bikash","timestamp":1627445263}
{"server_script_path":"/w","server_name":"www.wikidata.org","$schema":"/mediawiki/recentchange/1.0.0","minor":false,"bot":false,"wiki":"wikidatawiki","length":{"new":22936,"old":22819},"type":"edit","title":"Q80075234","revision":{"new":1468164258,"old":1467697544},"patrolled":true,"meta":{"dt":"2021-07-28T04:07:44Z","partition":0,"offset":363427334,"stream":"mediawiki.recentchange","domain":"www.wikidata.org","topic":"codfw.mediawiki.recentchange","id":"7016afae-6691-4dca-bfaf-a5a3363edf31","uri":"https://www.wikidata.org/wiki/Q80075234","request_id":"aa4f6828-149d-4feb-a3cf-cd39902773fe"},"namespace":0,"comment":"/* wbsetdescription-add:1|dv */ އަކުއިލާ ނަކަތުގައިވާ ތަރިއެއް, [[:toollabs:quickstatements/#/batch/60416|batch #60416]]","id":1514670484,"server_url":"https://www.wikidata.org","user":"EN-Jungwon","parsedcomment":"\u200e<span dir=\"auto\"><span class=\"autocomment\">Added [dv] description: <\/span><\/span> އަކުއިލާ ނަކަތުގައިވާ ތަރިއެއް, <a href=\"https://iw.toolforge.org/quickstatements/#.2Fbatch.2F60416\" class=\"extiw\" title=\"toollabs:quickstatements/\">batch #60416<\/a>","timestamp":1627445264}
{"server_script_path":"/w","server_name":"de.wikipedia.org","$schema":"/mediawiki/recentchange/1.0.0","minor":false,"bot":true,"wiki":"dewiki","length":{"new":17069,"old":17075},"type":"edit","title":"Liste der Biografien/Caro","revision":{"new":214271460,"old":213857611},"meta":{"dt":"2021-07-28T04:07:43Z","partition":0,"offset":363427335,"stream":"mediawiki.recentchange","domain":"de.wikipedia.org","topic":"codfw.mediawiki.recentchange","id":"6618b0ab-eadf-405a-a474-ec2ad9fef8bb","uri":"https://de.wikipedia.org/wiki/Liste_der_Biografien/Caro","request_id":"23181b86-03de-4153-ad99-e7e20e611ed6"},"namespace":0,"comment":"Bot: Automatische Aktualisierung, siehe [[Benutzer:APPERbot/LdB]]","id":309672385,"server_url":"https://de.wikipedia.org","user":"APPERbot","parsedcomment":"Bot: Automatische Aktualisierung, siehe <a href=\"/wiki/Benutzer:APPERbot/LdB\" title=\"Benutzer:APPERbot/LdB\">Benutzer:APPERbot/LdB<\/a>","timestamp":1627445263}

Here's the last entry, but formatted for better understanding:

{
  "$schema": "/mediawiki/recentchange/1.0.0",
  "bot": true,
  "comment": "Bot: Automatische Aktualisierung, siehe [[Benutzer:APPERbot/LdB]]",
  "id": 309672385,
  "length": {
    "new": 17069,
    "old": 17075
  },
  "meta": {
    "domain": "de.wikipedia.org",
    "dt": "2021-07-28T04:07:43Z",
    "id": "6618b0ab-eadf-405a-a474-ec2ad9fef8bb",
    "offset": 363427335,
    "partition": 0,
    "request_id": "23181b86-03de-4153-ad99-e7e20e611ed6",
    "stream": "mediawiki.recentchange",
    "topic": "codfw.mediawiki.recentchange",
    "uri": "https://de.wikipedia.org/wiki/Liste_der_Biografien/Caro"
  },
  "minor": false,
  "namespace": 0,
  "parsedcomment": "Bot: Automatische Aktualisierung, siehe Benutzer:APPERbot/LdB",
  "revision": {
    "new": 214271460,
    "old": 213857611
  },
  "server_name": "de.wikipedia.org",
  "server_script_path": "/w",
  "server_url": "https://de.wikipedia.org",
  "timestamp": 1627445263,
  "title": "Liste der Biografien/Caro",
  "type": "edit",
  "user": "APPERbot",
  "wiki": "dewiki"
}

Kibana for data visualization

As I mentioned in the introduction, we have a fantastic tool at our disposal for data visualization that doesn't require writing code, and that tool is Kibana. Kibana is part of the so-called ELK stack:

Elasticsearch provides the storage and indexing part
Finally, Kibana offers dashboards and widgets to explore and visualize data stored in Elasticsearch

Instead of writing to the standard output, we are going to write to an Elasticsearch instance. For that, we need to create the Sink. While you can use the Elasticsearch API directly, Hazelcast provides an extension to ease your job. Just add the com.hazelcast.jet:hazelcast-jet-elasticsearch-7 JAR to the classpath, and you can write the following:

private val clientBuilder = {
  val env = System.getenv()
  val user = env.getOrDefault("ELASTICSEARCH_USERNAME", "elastic")      // 1
  val password = env.getOrDefault("ELASTICSEARCH_PASSWORD", "changeme") // 1
  val host = env.getOrDefault("ELASTICSEARCH_HOST", "localhost")        // 1
  val port = env.getOrDefault("ELASTICSEARCH_PORT", "9200").toInt()     // 1
  ElasticClients.client(user, password, host, port)                     // 2
}

val elasticsearch = ElasticSinks.elastic(clientBuilder) {
  IndexRequest("wikipedia").source(it.toString(), XContentType.JSON)    // 3
}

Provide some parameterization to allow to run in different environments
Connect to the configured Elasticsearch instance
Effectively send the data to ES. Under the cover, Hazelcast will batch the requests.

Now, the pipeline can be improved:

val pipeline = Pipeline.create().apply {
  readFrom(wikipedia)
    .withTimestamps({ it.getLong("timestamp") }, 100)
    .writeTo(elasticsearch)
}
Hazelcast.bootstrappedInstance().jet.newJob(pipeline)

The icing on the cake, with good naming, the Hazelcast API allows people who are not developers to follow the logic along.

Running the above pipeline, we can already see the results in Kibana. In case you don't have an instance available, the GitHub repository provides a docker-compose.yml file. You only need to start the infrastructure with docker compose up.

Navigate to http://localhost:5601 with your favorite browser
Authenticate with login elastic and password changeme
Click on the "Create index pattern" button
Enter wikipedia for the index name
Click on the "Next step" button
Choose field for meta.dt for the Time field
Finalize by clicking on the "Create index pattern" button
On the left menu, select Analytics → Discover

You should see something like this:

Within this view, you can see all ingested documents. To explore further, you can choose the fields you want to see (on the right) and filter out documents based on their structure (in the Search bar above).

Curating "wrong" data

If you tried executing the job with the code at this step, you might have noticed that after some time, Elasticsearch stops ingesting data. Looking at the Hazelcast logs, you may notice a similar stack trace:

15:02:34.898 [ WARN] [c.h.j.i.e.TaskletExecutionService] [192.168.1.62]:5701 [dev] [5.0-BETA-1] Exception in ProcessorTasklet{068f-8bfa-4080-0001/elasticSink#0}
com.hazelcast.jet.JetException: failure in bulk execution:
[0]: index [wikipedia], type [_doc], id [PD017XoBfeUJ26i8qT-H], message [ElasticsearchException[Elasticsearch exception [type=mapper_parsing_exception, reason=object mapping for [log_params] tried to parse field [null] as object, but found a concrete value]]]
  at com.hazelcast.jet.elastic.ElasticSinkBuilder$BulkContext.lambda$flush$0(ElasticSinkBuilder.java:248)
  at com.hazelcast.jet.elastic.impl.RetryUtils.withRetry(RetryUtils.java:57)
  at com.hazelcast.jet.elastic.ElasticSinkBuilder$BulkContext.flush(ElasticSinkBuilder.java:244)
  at com.hazelcast.function.ConsumerEx.accept(ConsumerEx.java:47)
  at com.hazelcast.jet.impl.connector.WriteBufferedP.process(WriteBufferedP.java:73)
  at com.hazelcast.jet.impl.processor.ProcessorWrapper.process(ProcessorWrapper.java:97)
  at com.hazelcast.jet.impl.pipeline.FunctionAdapter$AdaptingProcessor.process(FunctionAdapter.java:226)
  at com.hazelcast.jet.impl.execution.ProcessorTasklet.lambda$processInbox$2f647568$2(ProcessorTasklet.java:439)
  at com.hazelcast.jet.function.RunnableEx.run(RunnableEx.java:31)
  at com.hazelcast.jet.impl.util.Util.doWithClassLoader(Util.java:498)
  at com.hazelcast.jet.impl.execution.ProcessorTasklet.processInbox(ProcessorTasklet.java:439)
  at com.hazelcast.jet.impl.execution.ProcessorTasklet.stateMachineStep(ProcessorTasklet.java:305)
  at com.hazelcast.jet.impl.execution.ProcessorTasklet.stateMachineStep(ProcessorTasklet.java:300)
  at com.hazelcast.jet.impl.execution.ProcessorTasklet.stateMachineStep(ProcessorTasklet.java:281)
  at com.hazelcast.jet.impl.execution.ProcessorTasklet.call(ProcessorTasklet.java:255)
  at com.hazelcast.jet.impl.execution.TaskletExecutionService$BlockingWorker.run(TaskletExecutionService.java:298)
  at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
  at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
  at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
  at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
  at java.base/java.lang.Thread.run(Thread.java:829)

It happens because of the way Elasticsearch works. As we didn't provide any explicit index schema, Elasticsearch inferred one for us from the first data payload it received. In this case, the log_params attribute has mostly the following structure:

"log_params": {
  "userid": 108038
}

Hence, Elasticsearch recognizes it as a JSON object with the userid property. Yet, sometimes, the stream contains "log_params":[], which is JSON array. Elasticsearch cannot reconcile between the two and throws the above exception.

To fix this, we can either filter out such data or transform the empty array property in an empty object property. As we would like to keep as much data as possible, let's choose the second option. As of now, we don't know if we will need to do it for another field, so it might be a good idea to make it generic:

class MakeFieldObjectIfArray(private val fieldName: String) : FunctionEx<JSONObject, JSONObject> { // 1
  override fun applyEx(json: JSONObject) = json.apply {           // 2
    if (json.has(fieldName) && json.get(fieldName) is JSONArray)  // 3
      put(fieldName, JSONObject())                                // 4
  }
}

val pipeline = Pipeline.create().apply {
  readFrom(wikipedia)
    .withTimestamps({ it.getLong("timestamp") }, 100)
    .map(MakeFieldObjectIfArray("log_params"))                    // 5
    .writeTo(elasticsearch)
}

Define a FunctionEx that takes a JSONObject as a parameter and returns a JSONObject
Return the same JSONObject with the following changes applied
If the object has a specific field and if this field is a JSONArray
Then replace the array with an empty JSONObject
Map each item in the pipeline using the previously-defined FunctionEx

Rerunning the pipeline now works without any glitch!

Making it more readable and "operable"

Because the pipeline is stable, it's time to refactor to build upon solid foundations. The refactoring goes along two axes:

Code readability, developer-oriented
"Operationability", ops-oriented

On the developer side, we can improve the experience by leveraging Hazelcast's API. GeneralStage offers the usual pipeline primitives: map(), flatMap(), filter() and a couple of more specialized ones. However, at this granularity level, we'd rather focus on the what instead of the how. For this reason, StreamStage also provides an apply() function that transforms a StreamStage into another StreamStage. Let's use it:

class MakeFieldObjectIfArray(private val fieldName: String) : FunctionEx<StreamStage, StreamStage> {
  override fun applyEx(stage: StreamStage) = stage.map { json ->
    json.apply {
      if (json.has(fieldName) && json.get(fieldName) is JSONArray)
        put(fieldName, JSONObject())
    }
  }
}

To use it:

val pipeline = Pipeline.create().apply {
  readFrom(wikipedia)
    .withTimestamps({ it.getLong("timestamp") }, 100)
    .apply(MakeFieldObjectIfArray("log_params"))       // 1
    .writeTo(elasticsearch)
}

Focus on the what

The next step is to improve the "operability". The only way to check what happens is to check Elasticsearch. If something happens in between (like above), it's hard to pinpoint exactly what the problem is. For that reason, we should add logging:

val pipeline = Pipeline.create().apply {
  readFrom(wikipedia)
    .withTimestamps({ it.getLong("timestamp") }, 100)
    .apply(MakeFieldObjectIfArray("log_params"))
    .peek()                                           // 1
    .writeTo(elasticsearch)
}

We log every item to the standard output

With a lot of data, this can be too much noise. A sample is enough:

fun sampleEvery(frequency: Int) = PredicateEx {
  Random.nextInt(frequency) == 0                      // 1
}

val toStringFn = FunctionEx<Any?, String> {
  it?.toString()                                      // 2
}

Return true if the random value between 0 and frequency is 0
Null-safe toString()

We can now put this code to good use:

val pipeline = Pipeline.create().apply {
  readFrom(wikipedia)
    .withTimestamps({ it.getLong("timestamp") }, 100)
    .apply(MakeFieldObjectIfArray("log_params"))
    .peek(sampleEvery(50), toStringFn)                // 1
    .writeTo(elasticsearch)
}

Sample one item per 50 on average

Also, Hazelcast provides an API to name each pipeline step.

class MakeFieldObjectIfArray(private val fieldName: String) : FunctionEx {
  override fun applyEx(stage: StreamStage) = stage
    .setName("remove-log-params-if-array")            // 1
    .map { json ->
      json.apply {
        if (json.has(fieldName) && json.get(fieldName) is JSONArray)
          put(fieldName, JSONObject())
      }
    }
}

With this, launching the pipeline outputs the following DAG log:

digraph DAG {
  "replace-log-params-if-array" [localParallelism=1];
  "replace-log-params-if-array-add-timestamps" [localParallelism=1];
  "map" [localParallelism=16];
  "elasticSink" [localParallelism=2];
  "replace-log-params-if-array" -> "replace-log-params-if-array-add-timestamps" [label="isolated", queueSize=1024];
  "replace-log-params-if-array-add-timestamps" -> "map" [queueSize=1024];
  "map" -> "elasticSink" [queueSize=1024];
}

Geo-locating data

Looking at the existing data, we can notice two types of contributions:

Contributions by authenticated users, e.g., GeographBot
Anonymous contributions, e.g.:
- 84.243.214.62, for IP v4
- 240D:2:A605:7600:A1DF:B7CA:5AF8:D971, for IP v6

There's no easy way to geo-locate using the former, but libraries and online APIs are available that leverage the latter. For this post, I've chosen to use MaxMind GeoIP database. It provides both a local file and a library to leverage it.

Let's add the necessary dependencies:

<dependency>
    <groupId>commons-validator</groupId>
    <artifactId>commons-validator</artifactId>
    <version>1.7</version>
</dependency>
<dependency>
    <groupId>com.maxmind.geoip2</groupId>
    <artifactId>geoip2</artifactId>
    <version>2.15.0</version>
</dependency>

Then, we can add an additional step in the processing pipeline to check whether the user is an IP and add the info if it is:

val enrichWithLocation = { stage: StreamStage ->
  stage.setName("enrich-with-location")                             // 1
    .mapUsingService(ServiceFactories.sharedService(databaseReaderSupplier)) { reader: DatabaseReader, json: JSONObject ->
       json.apply {
         if (!json.optBoolean("bot") && json.has("user")) {         // 2
           val user = json.getString("user")
           if (InetAddressValidator.getInstance().isValid(user)) {  // 3
             reader.tryCity(InetAddress.getByName(user))            // 4
                   .ifPresent { json.withLocationFrom(it) }         // 5
           }
         }
      }
    }
}

val pipeline = Pipeline.create().apply {
  readFrom(wikipedia)
    .withTimestamps({ it.getLong("timestamp") }, 100)
    .apply(MakeFieldObjectIfArray("log_params"))
    .apply(enrichWithLocation)                                      // 6
    .peek(sampleEvery(50), toStringFn)
    .writeTo(elasticsearch)
}

Set a descriptive name
If the bot property is false and if the user property exists
Validate that the user is an IP, v4 or v6
Geo-locate the IP
Add the data to the JSON
Add the step to the pipeline

Our first data visualization

With geo-located data, we would like to display changes on a world map. The good news, Kibana offers such a widget out-of-the-box.

Go to Analytics | Maps
Click on the Add Layer button
Select Documents
For the index, select wikipedia

Unfortunately, Kibana complains that the index doesn't contain any geospatial fields!

Indeed, while we formatted the data to create a data field that has latitude and longitude, Elasticsearch doesn't recognize it as a Geo-Point type. We need to map it explicitly. Worse, we cannot change the type of an existing field. Hence, we need to stop the pipeline, remove the current index, and lose all data.

Go to Management | Stack Management
Select Data | Index Management
Check wikipedia
Click on the Manage Index button
Select Delete index
Confirm deletion

We are now ready to map the field.

Go to Management | Stack Management
Select Data | Index Management
Click on the Index Template tab (the 3^rd one)
Click on the Create template button
Give it a relevant name, e.g., geo-locate
Set an index pattern that matches wikipedia, e.g., wikipedia
Click on Next to keep the defaults until you reach the 4^th step - Mappings
Add a new field named location.coordinates and with type Geo-point

Click again on Next until the last step. The preview tab should display the following JSON:

  "template": {
    "settings": {},
    "mappings": {
      "properties": {
        "location": {
          "properties": {
            "coordinates": {
              "type": "geo_point"
            }
          }
        }
      }
    },
    "aliases": {}
  }
}

Click on the Create template button

In the wikipedia index, Elasticsearch will map every field named coordinates inside a field named location to a Geo-Point. For that reason, we need to change the mapping code slightly.

Let's create such a dedicated mapping function:

private fun JSONObject.withLocationFrom(response: CityResponse) {
    val country = JSONObject()
        .put("iso", response.country.isoCode)
        .put("name", response.country.name)
    val coordinates = JSONArray()
        .put(response.location.longitude)
        .put(response.location.latitude)
    val location = JSONObject()
        .put("country", country)
        .put("coordinates", coordinates)
        .put("city", response.city.name)
        .put("timezone", response.location.timeZone)
        .put("accuracy-radius", response.location.accuracyRadius)
    put("location", location)
}

We can now use it in the pipeline:

reader.tryCity(InetAddress.getByName(user))
    .ifPresent { json.withLocationFrom(it) }

Let's start the pipeline again. Now, we can try to repeat the steps to create a Map. This time, it recognizes the field we mapped as a Geo-point and lets us go further.

Click on the Add layer button on the bottom right corner. You can already enjoy some data points displayed on the map.

Exploring data

Data points are excellent, but not enough. Suppose that we want to understand the entries by their location. For that, we need to add fields, i.e., meta.uri and comment. Don't forget to name the layer and save it. It's now possible to click on a data point to display the related data:

Wikipedia is a source of information for millions of users around the world. Because contributions can be anonymous (and remember those are geo-located), a malicious user can update an article not to benefit the community but to further a geopolitical agenda. We could ask the data whether the triplet article-language-location seems ok and does it raise some red flags. We already have the article via the meta.uri and the location, we need to add the language.

Adding derived data

Two main options are available to get the language:

From the server's URL, e.g., it.wikipedia.org implies Italian while fr.wikipedia.org mean French
From the comment (if it's not empty)

For educational purposes, I decided to go with the second one. Each event already contains a comment field. Here's a sample:

"Anagrames [[esgota]], [[esgotà]], més canvis cosmètics"
"Added location template"
"[[:傑克·威爾許]]已添加至分类，[[Special:WhatLinksHere/傑克·威爾許|此页面包含在其他页面之内]]"
"/* wbsetdescription-add:1|ru */ провинция Алжира"
"/* Treindiensten */"
"יצירת דף עם התוכן \"אסף \"בובי\" מרוז, יליד חיפה, הינו מוזיקאי, מתופף, חבר בלהקות אבטיפוס, קילר ואיפה הילד == הרכבים == === קילר הלוהטת === בשנת 1980 - שימש מתופף של הלהקה קילר הלוהטת. === אבטיפוס === הלהקה הוקמה ב[[קריות]] באמצע [[שנות השמונים]] ועברה גלגולי הרכב שונים. בגלגולה הראשון בשנת...\"
"Mooier wordt het er niet van."
"[[:Конуклар (Джиде)]] категори чу тоьхна"
Etc.

A linguist can infer the language of the field. It's also possible to use an automated process in the pipeline. A couple of NLP libraries are available in the JVM ecosystem, but I set my eyes on Lingua, one focused on language recognition.

We need to create an additional stage transforming function:

val languageDetectorSupplier = { _: ProcessorSupplier.Context ->
    LanguageDetectorBuilder
        .fromAllSpokenLanguages()
        .build()
}                                                                        // 1

val enrichWithLanguage = { stage: StreamStage<JSONObject> ->
    stage.setName("enrich-with-language")
        .mapUsingService(ServiceFactories.sharedService(languageDetectorSupplier)) { detector: LanguageDetector, json: JSONObject ->
            json.apply {
                val comment = json.optString("comment")
                if (comment.isNotEmpty()) {
                    val language = detector.detectLanguageOf(comment)   // 2
                    if (language != Language.UNKNOWN) {
                        json.put(
                            "language", JSONObject()                    // 3
                                .put("code2", language.isoCode639_1)
                                .put("code3", language.isoCode639_3)
                                .put("name", language.name)
                        )
                    }
                }
            }
        }
}

Create the function that provides the language detector
The magic happens here
Add language-related data to the JSON

We can now use the newly-defined function in the pipeline:

val pipeline = Pipeline.create().apply {
        readFrom(wikipedia)
            .withTimestamps({ it.getLong("timestamp") }, 100)
            .apply(MakeFieldObjectIfArray("log_params"))
            .apply(enrichWithLocation)
            .apply(enrichWithLanguage)
            .peek(sampleEvery(50), toStringFn)
            .writeTo(elasticsearch)
    }

On the Kibana map, you can now add any language related fields, e.g., language.name to display it along the rest of the data points. Yet, some of them have an empty comment field so that the language is not shown. One option would be to update the data pipeline accordingly, but it's also possible to filter out unwanted data points on the Kibana interface. In general, that's the way to go: push data anyway and leave what data they want to display to the final user.

On the map, go to the Filtering section and add a KQL filter that filters out data points with no value: language.name : *. The result is something like the following:

Refining data

It's already better, though we can notice some discrepancies:

"Ngfn" is not Sotho but more like somebody didn't find a good comment
"сюжет" can be Bulgarian, Kazakh, or Russian, definitely not Mongolian

Though Lingua has good results, it's fallible. Fortunately, Lingua can return a single language from a text snippet and a map of languages with a confidence rating. The first language has a confidence rating of 1.0; the other ones have a confidence rating between 0.0 and 1.0.

For example, comment "Nufüs" returns the following map:

1.0: Turkish
0.762256422055537: German
0.6951232183399704: Azerbaïjani
0.6670947340824422: Estonian
0.5291088632328994: Hungarian
0.36574326772623783: Catalan

Hence, the closest the confidence rating of the second language is to 1, the lower the confidence in the first language. To reflect that, we can add to the data point the difference between the 1.0 and the second language's confidence rating. The above code is updated as:

val languagesWithConfidence = detector.computeLanguageConfidenceValues(comment)             // 1
if (languagesWithConfidence.isNotEmpty()) {
    val mostLikelyLanguage = languagesWithConfidence.firstKey()
    val secondMostLikelyConfidence = languagesWithConfidence.filterNot { it.key == mostLikelyLanguage }.maxBy { it.value }?.value ?: 0.0 // 2
    json.put(
        "language", JSONObject()
            .put("code2", mostLikelyLanguage.isoCode639_1)
            .put("code3", mostLikelyLanguage.isoCode639_3)
            .put("name", mostLikelyLanguage.name)
            .put("confidence", 1.0 - secondMostLikelyConfidence)                                 // 3
    )
}

Get the sorted map of languages
Get the confidence rating of the second language, or 0 if the map has a single element
Add the confidence rating to the data point

Note that depending on the first data point of the pipeline, you might end up with a language.confidence field of type int, i.e., 0 or 1. If that happens, you need to delete the index and create an index template with a Double type as we already did above with the Geo-point.

At this point, you can display the language confidence and update the filter to filter out data points with low confidence, e.g., language.name : * and language.confidence > 0.2. Here's the result:

Conclusion

In this post, we have described how you could visualize and explore a data set with the help of the Hazelcast Platform for the pipelining part and Kibana for the visualization part. The latter doesn't need any front-end coding skills - or any coding skills whatsoever. You don't need to be a Pythonista nor a graphical library expert to start exploring your data sets now: being a developer on the JVM is enough.

Start exploring now!

hazelcast-demos / wikipedia-changes

Repository for the post and talk "Hazelcast + Kibana: best buddies for exploring and visualizing data"

New Release: Hazelcast Go Client v1.0.0

Yüce Tekol — Thu, 15 Jul 2021 11:18:57 +0000

We released Hazelcast Go Client v1.0.0.

This is the first major release of the revamped Go client. Of course the biggest news is, the client is compatible with Hazelcast 4.x and the upcoming Hazelcast 5.x. Here are the features included in this release:

Distributed Data Structures: Map, Replicated Map, Queue, Topic, List, Set, PNCounter,
Map aggregations,
Improved Map locks,
Load balancer,
External client public address discovery,
Lifecycle, cluster membership and distributed object listeners,
Smart routing,
Ownerless client,
JSON, Identified Data, Portable and Global serialization,
SSL connections (requires Hazelcast Enterprise),
Username/password credentials (requires Hazelcast Enterprise),
Hazelcast Management Center integration,
Hazelcast Cloud integration,
JSON configuration.

There are some missing features which exist in Go Client v0.6.0, such as the Near Cache, but we will complete them in the coming months, prioritized by the needs of our users.

Apart from Hazelcast 4.x+ compatibility, we aimed to implement conventions and best practices of the Go ecosystem. Here are some of the highlights:

Go context: Enables Go client to play well with other libraries that support the context package. Any client function call can be canceled manually or after a deadline. This was one of the most wanted features of the new Go client.
Go modules: Enables depending on a specific release without third party tools.
Idiomatic errors: The client library returns errors which can be easily unwrapped using the standard errors package.
Zero value configuration: The zero value of client configuration is the default configuration. Apart from being encouraged in idiomatic Go, it helps us provide a natural way of supporting JSON configuration.
Package based code layout: The client code is split into logical packages which collect relevant code together, such as cluster, aggregate, serialization, types and others.

Some useful links:

We are looking forward to feedback in order to provide the features our users want. Join us at our Slack channel at Hazelcast Community

Querying Your In-Memory-Data-Grid: Why and How?

Nicolas Fränkel — Thu, 13 May 2021 09:01:49 +0000

An in-memory data grid (IMDG) is a set of networked/clustered computers that pool together their random access memory (RAM) to let applications share data with other applications running in the cluster. Though IMDGs are sometimes generically described as a distributed in-memory data store, IMDGs offer more than just storage. IMDGs are built for data processing at extremely high speeds. They are designed for building and running large-scale applications that need more RAM than is typically available in a single computer server. This enables the highest application performance by using RAM along with the processing power of multiple computers that run tasks in parallel. IMDGs are especially valuable for applications that do extensive parallel processing on large data sets.

What Is an In-Memory Data Grid?

Where would you draw the line between a data store and a cache? Persistence? Hazelcast allows you to write your in-memory data on disk. Derived data vs. source of truth? If the cost of creating the data is cheap, why would you persist them? Let's agree that there's no clear-cut defining property but a blurry continuum between those two concepts.

The raison d'être for caches is two-fold: performance, data is available faster, and availability, data is "always" available (at least the availability is higher than the underlying store). Therefore, it stands to reason queries would benefit from the same properties. You may argue that caches are key-value stores: they shine when retrieving an entry by its key. However, if you execute a query, you'd need to iterate entry by entry, something akin to a full table scan in a SQL database. The performance would be abysmal, even more so if your cache is distributed like Hazelcast.

And yet, sometimes, the underlying store offers no straightforward query mechanism either. Think about a Kafka topic or a web service.

In this post, I'd like to describe how you can take advantage of Hazelcast to query your cached data in different ways and still be fast.

Our Use Case: Collecting User Interactions

Collecting user interactions with the application is the first step when you want to improve your application. Whether it's documentation to make it more relevant or e-commerce to increase your sales, so-called "clickstreams" can be seen as the new oil. We will use a (significantly) simplified click-collecting application architecture:

A web page with JavaScript code that sends click events to the server
The server part is a Python Flask application. It enriches the events and stores them in a Hazelcast cluster.
The Hazelcast cluster itself. It provides querying capabilities.
A third-party component to query the cache data.

In a real-world setup, we probably would want to store the event data in a persistent store, e.g., a Kafka topic or a MongoDB instance. We would then capture the inserts via a Jet pipeline to load the Hazelcast cluster. In the context of this post, it would only make the architecture unnecessarily complex. We can tentatively model the structure of an event with the following attributes:

The timestamp of the event
The date on the client machine; it might differ from the previous data.
The type, e.g., click, change, etc.
The screen coordinates
The source component ID
The IP of the client machine
The related session
Anything else that might be relevant

When users interact with the application, the front-end will send a JSON payload that contains client-related data to the server. The server will also enrich the JSON with server-related data, e.g., IP and session ID, and store the complete JSON in a Hazelcast IMap. The key is not that important; the value is the JSON. Now, our marketing team wants to query the data to understand how users interact with our application.

The Predicate API

The Predicate API is the one that predates all other ways presented in this post. The idea behind it is that it will serve as a filter before returning the values of an IMap. Before going further, we need to understand what it means to query a distributed system. From the documentation:

The requested predicate is sent to each member in the cluster. Each member looks at its own local entries and filters them according to the predicate. At this stage, key/value pairs of the entries are deserialized and then passed to the predicate. The predicate requester merges all the results coming from each member into a single set.

That being said, it's time to query! Here's the code to select all values that match a component's name:

var entry = Predicates.newPredicateBuilder().getEntryObject();
var predicate = entry.get("component").equal(name);             // 1
Collection<String> values = map.values(predicate);              // 2

name is the component's ID
values contains all values that match the component's name

The above is Java code, but all of our clients do offer the Criteria API. Unfortunately, if you run the code as-is, values will be empty even though some values match. The reason is that by default, Hazelcast stores values as bytes arrays. Thus, it has no understanding of the underlying format. The fastest way to fix the issue is to wrap the JSON value into a HazelcastJsonValue object. On the "put" side, we need to replace the first line with the second one:

analytics.set(uuid.uuid4(), json.dumps(data)).result()          # 1
analytics.set(uuid.uuid4(), HazelcastJsonValue(data)).result()

data is a Python dic

On the query side, the only change is to update the generic type of the collection:

Collection<String> values = map.values(predicate);
Collection<HazelcastJsonValue> values = map.values(predicate);

The Shape of Water Stored Data

Using HazelcastJsonValue means we are still storing data as JSON-formatted Strings. While some language ecosystems favor the usage of JSON, e.g., JavaScript, some others favor the usage of dedicated data structures, e.g., Java and Go. If developers on different language stacks use your Hazelcast cluster, you need to store the data in the most "consumable" shape. Enters PortableAnother data serialization format that is query-friendly. With Portable, you need to

Design a class or a struct (for Go) that models a unit of data.
Write down how to store and load an instance in an IMap

Let's keep the storage part in Python and the query part in Java. Here's the Python code:

from hazelcast.serialization.api import Portable

class Analytics(Portable):
    def __init__(self, dic=None):
        if not dic:
            return
        self.timestamp = dic['timestamp']
        self.event = dic['event']
        self.instant = dic['instant']
        self.component = dic['component']
        self.session = dic['session']
        self.client = dic['client']
        self.value = dic.get('value', None)
        self.type = dic.get('type', None)
        self.x = dic.get('x', None)
        self.y = dic.get('y', None)

    def get_class_id(self):
        return 1

    def get_factory_id(self):
        return 1

    def write_portable(self, writer):
        writer.write_long('timestamp', self.timestamp)
        writer.write_long('instant', self.instant)
        writer.write_utf('event', self.event)
        writer.write_utf('component', self.component)
        writer.write_utf('session', self.session)
        writer.write_int('client.left', self.client['left'])
        writer.write_int('client.height', self.client['height'])
        writer.write_int('client.top', self.client['top'])
        writer.write_int('client.width', self.client['width'])
        if self.value is not None:
            writer.write_utf('value', str(self.value))
        if self.type is not None:
            writer.write_utf('type', self.type)
        if self.x is None:
            writer.write_int('x', -1)
        else:
            writer.write_int('x', self.x)
        if self.y is None:
            writer.write_int('y', -1)
        else:
            writer.write_int('y', self.y)


factory = {
    1: Analytics
}

On the Java side, we can avoid creating the class by using GenericRecord. As its name implies, it's generic. Hence, we can query the data without having the definition of a Java Analytics class on the classpath! Again, the change is straightforward:

Collection<HazelcastJsonValue> values = map.values(predicate); 
Collection<GenericRecord> values = map.values(predicate);

In both cases, you can access a field by its name:

values.stream()
        .map(value -> value.getString("component"))
        .forEach(System.out::println);

Indexing Data

Indexing data is not (only!) related to databases and how one stores data on disk. If you already know about indexes, please feel free to skip this introduction. Key-Value stores, such as Hazelcast's IMap, provide fast key-based access: that's their raison d'être. However, to query the value - or part of it, the engine needs to:

Go through each of the entry in turn
Check whether the predicate applies
And eventually, add the entry to the collection of matching entries

While it's faster to execute the flow in-memory than reading from a disk, it's not optimal. The solution is to create a dedicated data structure that stores additional information: an index. You can imagine an index as a sort of Key-Value store. The key is the value of the attribute, and the value the collection of keys from entries whose attribute's value matches the value.

Now, when we query for entries whose value match "foo", the engine will look at the index, locate the key "foo", read the value, and look for entries whose keys are "a", "c" and "e". Of course, this is a pretty simple example. Since Hazelcast allows you to store complex values such as JSON and Portable, you can set an index on any attribute (or nested attribute).

Here, you'd be able to set an index on the "comp" attribute or the "t" attribute. You should carefully evaluate on which attribute to set an index, though, as indexes have two main downsides:

An index slows down storing values as the engine needs to maintain the index.
An index-only makes sense if a query is using it. With the above IMap, if the predicate is entry.get("component").equal("button"), but the index is on "t", it won't improve performance.

The above explanation caters to exact matches. Imagine that we want now every entry whose "t" is above a certain threshold, e.g., entry.get("t").lessThan(2). For this usage, we need a sorted index. A sorted index keeps its keys, well, sorted. The engine will go through the keys in order, and once it has reached a key that doesn't match the predicate, it will return the results. Finally, for queries that involve multiple attributes, we need a compound index that contains all attributes. Here's such a query:

var map = client.getMap("analytics");
var entry = Predicates.newPredicateBuilder().getEntryObject();
var predicate = entry.get("component").equal(name).and(entry.get("instant").lessThan(instant));

The relevant index configuration would be:

hazelcast:
  map:
    analytics:
      indexes:
        - type: HASH
          attributes:
            - "component"
        - type: SORTED
          attributes:
            - "instant"

At this point, we know how to query an IMap with the Criteria API and make it fast with the proper index configuration.

SQL and JDBC

As a Java developer, you might already be familiar with SQL and JDBC. Learning a new API might feel like a roadblock. The good news is that Hazelcast now offers a JDBC driver. To use it:

Add the driver to your classpath (here with Maven)

<dependency>
    <groupId>com.hazelcast</groupId>
    <artifactId>hazelcast-jdbc</artifactId>
    <version>4.2</version>
</dependency>

Use the standard JDBC API:

try (var connection = DriverManager.getConnection("jdbc:hazelcast://localhost:5701/");
    var statement = connection.prepareStatement("SELECT * FROM analytics where component = ? and instant < ?")) {
    statement.setString(1, name);
    statement.setLong(2, instant);
    var resultSet = statement.executeQuery();
    var modelBuilder = new TableModelBuilder<>();
    while (resultSet.next()) {
         var component = resultSet.getString("component");
         System.out.println(component);
    }
}

That's all! The icing on the cake, you can use SQL in other language stacks (without JDBC, obviously). At the time of this writing, it still uses part of the Criteria API, but we intend to move it outside. The above code can be rewritten in Python like the following:

analytics = client.get_map('analytics')
select = 'component = {component} AND instant < {instant}'.format(component = component, instant = instant)
results = analytics.values(sql(select)).result()
for result in results:
    print(result.component)

Conclusion

In this post, we went through several core concepts covering querying your data on Hazelcast:

It makes sense to query your in-memory data because data access is fast
By default, Hazelcast offers the Criteria API; you can use SQL as an alternative
Use indexes judiciously to speed up your queries

The complete code for this post is available on GitHub:

hazelcast-demos / query-cache

Demos different ways to query a Hazelcast IMap

Hazelcast, from Embedded to Client-Server

Nicolas Fränkel — Fri, 16 Apr 2021 08:31:16 +0000

Java developers are particularly spoiled when using Hazelcast. Because Hazelcast is developed in Java, it's available as a JAR, and we can integrate it as a library in our application. Just add it to the application's classpath, start a node, and we're good to go. However, I believe that once you start relying on Hazelcast as a critical infrastructure component, embedding limits your options. In this post, I'd like to dive a bit deeper into the subject.

Starting with embedded

As mentioned above, the easiest way for Java developers to start their journey using Hazelcast is to embed it in their application like any other library. During the application startup lifecycle, we just have to call Hazelcast.newHazelcastInstance(): this will start a new Hazelcast node in the currently running JVM. Because of Hazelcast auto-discovery capabilities, without further configuration, nodes will discover each other and form a cluster. In a couple of minutes of development time, we can create a distributed In-Memory Data Grid. Hard to do better in terms of Developer Experience!

Spring Boot makes our life even easier. Just add a hazelcast.xml file at the root of the classpath (or a Config bean to the application context), and the framework will start Hazelcast for you with the given configuration.

It's a great starting point, but I hope that at some point, your application is going to receive more attention, which will increase the load. To handle it, a couple of solutions are available:

Keep the same number of nodes but try to configure the JVM for better performance
Scale horizontally, i.e., add more nodes to the cluster.
Scale vertically, i.e., give each node more resources.

In all of those cases, the coupling of the application and Hazelcast limits the benefits. For example, you'll need to choose to configure the JVM for the application workload or Hazelcast's. You'll need to scale horizontally according to the most limiting factor, the app or Hazelcast. And by scaling vertically, there's no way to assign the additional resources to one or the other facet.

To avoid those problems, the next logical step is to migrate to a client-server deployment model.

Client-Server Deployment

The principle underlying Client-Server deployment is the separation of concerns: decoupling the application from the Hazelcast node.

The benefits are readily available:

We can configure the Hazelcast JVM in relation to its workload profile only
We can scale the Hazelcast cluster horizontally independently from the application's
We can allocate resources to either Hazelcast nodes or the application's, and they won't compete for them

As an additional benefit, we can now use any (or all!) other languages that Hazelcast provides: C++, Python, Node.js, Go, C#, and of course, Java.

Unfortunately, there's no such thing as a free lunch. While moving from embedded to client-server makes it possible to scale quickly, new pitfalls have cropped up.

Pitfall #0: Launching

The first pitfall is not a true one but a simple question. How do we launch Hazelcast now that it's not embedded anymore?

The easiest way to launch Hazelcast is to use the command line. Install it. Run it. It's the fastest path to get you started, but this is not recommended for production. Alternatively, you can also download the Hazelcast distribution and run the start script. The command line is intended for development purposes: it's not fit for production usage unless you build a custom operations layer on top - and we don't recommend it.

If you're using containerization technologies, we offer Docker images. If you run your workload on Kubernetes, check our Helm chart.

This section is only a summary. Please be sure to check the documentation for all available options and their respective details.

Pitfall #1: Classpath

In embedded mode, the JVM runs both the application and Hazelcast. If you want to add any of the provided extension points, you just need to register it, and it will work out of the box as the application and Hazelcast share the same classpath: this is not true anymore for client-server.

Let's detail a concrete example. Imagine that our application uses Hazelcast to cache data from a database. We want to decouple our architecture and, to do so, implement a Read-Through/Write-Through cache. This way, the application only interacts with the cache, while the cache interacts with the database. Hazelcast makes it possible with the MapLoader interface for Read-Through and MapStore for Write-Through.

We could implement those interfaces in our project and the implementation classes via the configuration in embedded mode. But now, we need to create a dedicated Java project with the Hazelcast dependency, implement the interfaces in this project, package the project as a JAR, and add it to the classpath of each of the running Hazelcast nodes.

While it's neither complex nor complicated, it's additional work.

Another related classpath management issue comes from domain classes. On the JVM, with the notable exception of Clojure, developers tend to model domain entities with classes, e.g., Person, Customer, User, etc. If the above extensions use those classes, e.g., the MapLoader reads from the database, creates a new Person instance, and stores it in the IMap, they also need to be on the members' classpath.

Because they're bound to the business, domain classes probably change more often than classes that customize Hazelcast. Whenever a domain class changes, we need to update the JAR and restart each node with the updated JAR version. Worse, different nodes will run different versions. If the change is not forward-compatible (a very likely occurrence), exceptions will pop here and there. Hence, we need to shut down the whole cluster before upgrading.

To cope with that, we have two solutions:

The legacy path is user code deployment. It allows to "send" code from a client member to the cluster. This way, no restart is needed. On the flip side, the cluster needs to accept that an external component can change the code it runs. It extends the attack surface of your system. With user code deployment, you're trading less security for more agility. Therefore, you need to weigh the pros and cons before walking this path.
With Hazelcast 4.1, we provide an alternative in the form of generic records. As its name implies, such a record is generic, meaning you can access the data without the class on the member's classpath. Generic records come with some limitations as well. For example, while you can read data, you cannot write it. Note that generic records are considered BETA at the time of this writing.

Pitfall #2: Serialization

While running embedded, domain objects are straightforward: implement Serializable without a second thought and "it just works." If you migrate to a client-server deployment, provided that you curate the members' classpath as mentioned above, it will work as well. Note that even in that case, Java serialization is not very performant, wastes space, and is a security concern.

However, the client-server deployment model opens new doors: the cluster data is suddenly more visible. Other developers, teams, or departments in your organization might be interested in using this newly-found source of data. But maybe you don't want to couple your teams by providing them a JAR of your domain classes so they can deserialize the domain objects? Or perhaps their technology stack might not be based on the JVM, so they cannot deserialize the data at all? In all cases, it opens an interesting debate regarding which format you should use to store data.

Hazelcast offers a couple of out-of-the-box formats and can integrate with virtually anything. But in embedded mode, we limit ourselves unnecessarily. Imagine a Hazelcast cluster that needs to serve both Java clients and Python clients. Each stack manages items with its class representation.

The Portable serialization mechanism allows us to store data in one stack and to access it from the other - and the other way around. Its main benefit is that it doesn't require reflection. Even if you don't use multiple languages, you can not only store the representation of your data on the server-side without having the classes on the members' classpath, but you can also query it. Did you have a look at our new SQL API by the way? Given that business classes' lifecycle is bound to be short, it's one less concern during deployment.

Pitfall #3: Embed During Development, Client in Production

The final problem is how to manage Hazelcast during the development process.

In Java, it's unnecessary to embed a node during development and run it as a client during production. It can be achieved in different ways:

Hazelcast itself allows you to override properties individually and to point to a specific configuration file. It's even possible to compose different configuration snippets.
Alternatively, if you're using Spring Boot, it's possible to provide different configurations depending on a profile. Spring Profiles are a runtime feature. Please check the relevant documentation for more information on how to create profiles.
The Maven build tool also offers a profile feature, but at build-time. While it's an option, my experience is that build-time configuration can increase the number of artifacts.
Finally, you can combine the options above.

Those are great options if you're developing in Java, but all is not lost if you aren't. Containerization can help us a lot! For example, the TestContainers project is available in all stack that Hazelcast support: Python developers can easily leverage the Python project to set up a local Hazelcast cluster quickly, Go developers the Go project, C# developers the .Net project, etc.

Conclusion

While it's easier to start Hazelcast if you're a Java developer, the embedded deployment model is limited compared to the Client-Server one. Migrating to the later deployment model allows you to "free" your data and make it available across the organization.

You'll encounter the pitfalls mentioned above along the way. It's perfectly normal. I hope that this post will allow you to overcome them easily.

Beyond "Hello World": Zero-Downtime Deployments on Kubernetes

Nicolas Fränkel — Thu, 04 Mar 2021 16:07:21 +0000

Today, I'd like to offer an approach that will allow most organizations to proceed with zero-downtime deployments with help from the Hazelcast platform.

Applications are not stateless!

A couple of years ago, I became interested in Continuous Deployment as the natural extension of my deep interest in Continuous Integration. Among the CD patterns, Blue-Green deployment was the first. Here's a sample architecture:

The idea is that a router component directs the requests to one environment; by convention blue. This environment contains all the necessary components for the application to achieve its task. Meanwhile, Ops can deploy a new version of the application in the green environment. When all is ready, they can switch the router to direct all the requests to the new green environment.

Unfortunately, this ideal scenario suffers from several drawbacks, all related to the state. Though we would like our applications to be stateless, most of them are not. Let's chase the devil where it lies.

Sessions (and cookies) are ubiquitous in nearly all web applications. In general, developers use sessions when data are both is small and required frequently. When that happens, fetching data from the database is too expensive. A simple use-case of session usage is to display the logged-in user's name on each screen in a web app.

Problems start to happen when the node which hosts the session dies. The data is lost. I believe this is a solved problem now, thanks to session replication. By the way, did you know that Hazelcast has many different session replication approaches: Jetty, Tomcat, WebFilter-based, and even Spring Sessions?

You should store data that don't fit the session use-case in external databases. It looks like it solves the issue from a bird's eye view, but it doesn't. You need to migrate data in the blue environment when you switch to the green environment. In real-world applications, it's bound to take time.

Imagine an e-commerce web application where we store users' carts in the database. When the switch from blue to green starts, we need to move users to the latter. Their session is migrated to the green application server, which uses the green database. We somehow need to migrate the data as well. Unfortunately, we face a quandary: we cannot redirect the user before migrating the data, but we cannot migrate the data while still using the blue environment.

The obvious escape route is to:

Direct new users to the green environment
Wait until an existing user has stopped using the application (by checking session expiries) and then move their data
When all user sessions on blue have expired, transfer the remaining data

This process is lengthy and error-prone. It won't be feasible to migrate several times a day.

A tentative solution, the shared database

The most straightforward solution to cope with the above problem is not to migrate but to use a database (or cluster) shared between the two environments.

I wrote about this approach in the past. Please find here the abridged version.

The biggest downside of sharing the database is that some application deployments require a database schema change. But some changes are not backward-compatible. Hence, because we need to be able to roll back the deployment while keeping the data, we need to split a breaking change into a series of side-by-side compatible changes. This way, version x.0 of the application could still work with version x.1 of the database schema.

This holds whether the database enforces schema-on-write (as most SQL databases do) or the application needs to apply schema-on-read (à la NoSQL).

The split of changes makes the deployment process manual and fragile. Moreover, we still need to migrate data from one form to another, even in the same database. We can achieve it through different channels: the application code that touches the data, triggers, etc.

The final nail in the coffin boils down to how our process manages. It manages the "hot" ones, but what about the "cold" ones? As in the previous approach, data that isn't touched during the migration process still needs to be migrated.

Embracing data migration

Whether we design the deployment architecture around a database in each environment or a single shared one, we still need to migrate the data anyway. Hence, it's much better to embrace migration instead of pushing it aside and treating it as a second-class problem.

The problem behind data migration is not the migration itself but how we manage it. Batch jobs are the standard way to implement data migration. From a very general perspective, one plans a batch to run at specified intervals, read data that meets specific criteria from a source, transform the data, and finally write the transformed data into a sink. When the job has finished its task, it stops.

Batches come with several drawbacks:

What if data has not yet arrived when the job finishes?
How to hold the whole data to process in memory?
What if the batch fails mid-way?
What if the batch runs so long it oversteps its next schedule, e.g., an hourly batch that takes more than 60 minutes?

The good thing is that the batch model has been with us so long that we have devised ways to cope with those drawbacks. For example, we can handle the data in chunks, keeping a cursor to the last processed chunk to restart from this point if we need to.

The only way to cope with the "data has not yet arrived" problem is to process it at the next run, which may be in the next hour. For data migration, this is not possible. We don't want to wait another hour to process the latecomers' data and finish the migration. The issue stems from the bounded nature of the batch model. It starts... and it ends.

What if we started a never-ending job and it would process data as soon as it comes in?

Stream processing and Change-Data-Capture to the rescue

This never-ending approach has a name: stream processing (or data streaming). Stream processing answers all of the previous drawbacks of the batch approach through an event-driven paradigm.

What if data has not yet arrived when the job finishes: the job never completes; it processes data when they become available
How to hold the whole data to process in memory: data arrive in small chunks, so the memory consumption of one data item is relatively small
What if the batch fails mid-way: most stream processing engines, such as Hazelcast Jet, are distributed
What if the batch runs so long it oversteps its next schedule: one doesn't schedule stream processing jobs; they can run forever until you stop them

We need a way for the database to emit such data events. Fortunately, it exists through Change-Data-Capture and one of its implementation, Debezium. In short, Debezium works by reading the append-only log file that SQL databases leverage to provide fail-over. When the primary node receives a write statement, it writes it down into this append-only log. Both the primary and all secondary nodes read that file and apply the statement. This way, they always maintain the same state, and if the primary node fails, one of the secondary nodes can take over from the same state.

Hazelcast Jet leverages Debezium as a library to read the append-only log.

To read more about the details on how Hazelcast Jet and Debezium, please refer to the documentation.

A blue-green deployment on Kubernetes in 6 steps

Now that we have all the pieces in place, we can finally start designing the solution. I'll be using Kubernetes because it's a pretty popular platform, but there's nothing Kubernetes specific about the solution. It could run on any infrastructure.

Note that when one reads about Kubernetes, it's easy to think that its rolling upgrade feature allows for seamless continuous deployment of containers. That's correct, up to a point. While it's a no-brainer to deploy stateless applications, it doesn't concern itself with the state. We are the ones to take care of it.

The starting point for our blue-green deployment is an application in a Kubernetes Deployment with a replica count of 3 and a RollingUpdate strategy. They all use the same database pod located in the existing blue environment. We configure all application pods with session replication.
The next step is to schedule the new database pod in the now-empty green environment. It can take as long as necessary; the application pods won't use it until later.
This the critical step: we schedule the data migration job pod (labeled forward in the next diagram). As soon as an application pod executes a write to the blue database pod, the forward job captures it, reads its data, transforms them according to the new schema in the green database pod, and writes them there. Hence, while the blue database is the source of truth, you can consider the green one as derived data. Moreover, it's kept in sync with the former - not in real-time, as even light doesn't travel instantaneously, but close enough for our needs.
Now is the time to schedule the new version of the application. We just tell Kubernetes about the new image in the Deployment. It's its job to achieve that via the rolling update. With session replication and both databases closely in sync, we can seamlessly move users from a v1 app pod that uses the blue database pod to a v2 app pod that uses the green database pod.
At some point, all users will have migrated from the blue environment to the green environment. At this point, no writes happen in the blue environment. The green environment has become the active one.
It's time to do some cleanup. We can stop the forward job and unschedule its pod(s). Later, we can do the same with the blue database pod.

Note that for the sake of simplicity, we described only the nominal path. In real-world scenarios, you should prepare a backward job that reads from the green database and writes back to the blue one. If you need to rollback, you'll stop the forward job, schedule the backward job, and downgrade your application pods again to move users back to the original blue environment.

Conclusion

In this post, I described the core problem of upgrading an application: state. The state is in two places, applications themselves with user sessions and databases. Session replication is a solved problem, database migration not so much.

With data streaming and Change-Data-Capture, we have a foolproof process to migrate data from databases to update an application while serving users at the same time.

Please find the associated Git repository that illustrates the post:

hazelcast-demos / zerodowntime

Demo notes

Set up

Create a kind cluster with correct port mapping

kind create cluster --config kind.yml

Set the Kubernetes context

kubectl config set-context --current --namespace=zerodowntime

Get back to the v1 of the application

git checkout 1.0

Build the image: The build will load the image to the Docker daemon.

./mvnw clean compile com.google.cloud.tools:jib-maven-plugin:dockerBuild

Load it into kind.

kind load docker-image hazelcast/hzshop:1.0

Get to the v2 of the application

git checkout master

Build the image and load it to the Docker daemon

./mvnw clean compile com.google.cloud.tools:jib-maven-plugin:dockerBuild

Load it into kind.

kind load docker-image hazelcast/hzshop:2.0

Load the remaining images in kind

kind load docker-image mysql:8.0
kind load docker-image hazelcast/forward:2.0

Execution

Schedule common parameters

kubectl apply -f infrastructure/kube/parameters.yml

Schedule the original database (blue)

kubectl apply -f infrastructure/kube/blue.yml

Schedule the v1 version of the application

kubectl apply -f infrastructure/kube/application.yml

Use the application to add products to the cart of any…

View on GitHub

Hazelcast Command Line is released!

Alparslan Avcı — Wed, 24 Feb 2021 13:37:41 +0000

We are pleased to announce that Hazelcast Command Line 4.2020.12 is now available. It supports Hazelcast IMDG 4.1.1 and Hazelcast Management Center 4.2020.12. It took longer than anticipated to get a proper CLI for Hazelcast, however, we hope you will find it was worth the wait.

In this blog post, we will go through what problems we aim to solve with Hazelcast CLI, as well as how to use it. If you're more of a hands-on learned, feel free to skip directly to the end for tips on using it.

Installing and starting an Hazelcast instance, the long way

As you most likely know, you can use Hazelcast in two main topologies: Embedded and Client-Server. You add Hazelcast as a dependency to your Java application if you choose the former architecture.

But, if you would like to design your architecture using a client-server approach, you need to start Hazelcast member instances. Before the Hazelcast CLI, you had to perform multiple steps to run a Hazelcast member standalone.

Install Java

Since Hazelcast member is a Java application, it requires a JRE to be installed in the environment. If you are not familiar with Java, it might be difficult to find the proper version and install it.

Download Hazelcast distribution

Hazelcast is available for download at https://hazelcast.com/get-started/. You had to visit this page and download the Hazelcast version distribution that you would like to install.

Run the instance using start.sh script

After downloading the distribution, you had to extract it to a proper location and then run the “start.sh” script that resides in the “bin/” directory.

After all these steps complete successfully, you (finally) have a running Hazelcast member instance.

However, this method is a manual process. There is support for neither package managers nor version control. Moreover, installing Hazelcast Management Center requires the same steps again.

The easy way: using Hazelcast Command Line

To make the Hazelcast installation easier, we developed Hazelcast Command Line. Please note that it is only targeted at developers and not recommended for usage in production environments for the time being. Now, you only need to use your favourite package manager to install Hazelcast, and then run the hz start command to start a Hazelcast member instance!

Even better, we have deployed the CLI in the most common repository managers.

Install with Homebrew

To install with Homebrew, you first need to tap the hazelcast/hz repository. Once you’ve tapped the repo, you can use brew install to install:

$ brew tap hazelcast/hz
$ brew install hazelcast

Install with yum/dnf

The RPM packages for Hazelcast Command Line are kept in Hazelcast's RPM repository. Just run the following commands to install it using yum/dnf:

$ wget https://repository.hazelcast.com/rpm/hazelcast-rpm.repo -O hazelcast-rpm.repo
$ sudo mv hazelcast-rpm.repo /etc/yum.repos.d/
$ sudo yum install hazelcast

Install with apt

You can find the Debian packages for Hazelcast Command Line in Hazelcast's Debian repository. Run the following commands to install it using apt:

$ wget -qO - https://repository.hazelcast.com/api/gpg/key/public | sudo apt-key add -
$ echo "deb https://repository.hazelcast.com/debian stable main" | sudo tee -a /etc/apt/sources.list
$ sudo apt update && sudo apt install hazelcast

All these package managers will handle the Java dependencies for you. Also, you can pick the previous versions by using the proper commands for each package manager.

After the installation, run the following command to see everything is completed successfully:

$ hz
Hazelcast IMDG 4.1.1
Usage: hz [-hV] [COMMAND]
Utility for the Hazelcast operations.

Global options are:

  -h, --help Show this help message and exit.
  -V, --version Print version information and exit.
Commands:
  start Starts a new Hazelcast IMDG member
  mc Utility for Hazelcast Management Center operations.

A quick intro to Hazelcast CLI

The main command of Hazelcast Command Line is hz. To start a Hazelcast instance, just run hz start and a new instance will run with the default configuration:

$ hz start
Aug 21, 2020 1:40:04 PM com.hazelcast.config.FileSystemYamlConfig
INFO: Configuring Hazelcast from '/Users/myuser/hazelcast-command-line/distro/build/dist/config/hazelcast.yaml'.
...

If you want to configure the instance with an external Hazelcast configuration file, use -c option with the command as below:

$ hz start -c /full/path/to/config-file.yaml

The -h option reveals further available options:

$ hz start -h
Usage: hz start [-hvV] [-vv] [-c=<file>] [-i=<interface>] [-p=<port>] [-j=<path>
                [,<path>...]]... [-J=<option>[,<option>...]]...
Starts a new Hazelcast IMDG member
  -h, --help Show this help message and exit.
  -V, --version Print version information and exit.
  -c, --config=<file> Use <file> for Hazelcast configuration. Accepted
      formats are XML and YAML.
  -p, --port=<port> Bind to the specified <port>. Please note that if the
      specified port is in use, it will auto-increment to
      the first free port. (default: 5701)
  -i, --interface=<interface>
      Bind to the specified <interface>.
  -j, --jar=<path>[,<path>...]
      Add <path> to Hazelcast classpath (Use ',' to separate
      multiple paths). You can add jars, classes, or the
      directories that contain classes/jars.
  -J, --JAVA_OPTS=<option>[,<option>...]
      Specify additional Java <option> (Use ',' to separate
      multiple options).
  -v, --verbose Output with FINE level verbose logging.
  -vv, --vverbose Output with FINEST level verbose logging.

As we mentioned above, you can also start a Hazelcast Management Center using Hazelcast Command Line. Please run the following command to start a Hazelcast Management Center instance:

$ hz mc start

If you would like to start a Hazelcast Management Center with custom context path and port, please run the following command:

$ hz mc start -c [new-context-path] -p [port]

Future

Currently, Hazelcast CLI is only targeted at developers and not recommended for usage in production environments. Also, there is no support for Windows environments. We are working on them and will include more features in future releases.

Your feedback as a user is very valuable to us. We would appreciate your thoughts and feedback with us by creating an issue on Hazelcast Command Line Github repository.

Hazelcast Discovery Auto Detection

Rafał Leszko — Fri, 08 Jan 2021 15:56:27 +0000

Hazelcast was always well integrated with all cloud environments thanks to discovery plugins like Hazelcast AWS Plugin or Hazelcast Kubernetes Plugin. With the IMDG 4.1 release, however, we went one step further in making Hazelcast as user friendly as possible. Hazelcast can now automatically discover the environment it is running in and find a suitable discovery plugin. What’s more, we made it all extensible, so if you develop any discovery plugin by yourself, you can make your plugin auto-discoverable.

In this blog post, I walk through common cloud environments and explain how auto-detection works in each of them. You can also check Hazelcast Reference Manual: Discovering Members by Auto Detection to find more about the requirements for the discovery auto-detection.

AWS Auto Detection

To start Hazelcast on AWS EC2 Instances, you have to perform the following steps for each instance:

Configure Security Groups to open port 5701 (the default Hazelcast port)
Configure IAM Role assigned to EC2 Instance to allow the ec2:DescribeInstances permission
Install Hazelcast CLI

Now, execute the following command on each of the EC2 Instances.

hz start
...
Auto-detection selected discovery strategy: class com.hazelcast.aws.AwsDiscoveryStrategyFactory
...
Members {size:2, ver:2} [
        Member [172.31.14.177]:5701 - 383a7f9e-e18c-4646-a998-d12810a677b8
        Member [172.31.3.163]:5701 - 956cf0cb-24d8-4604-9ad3-1a4785d00a3b this
]

That’s it! Hazelcast automatically detected that it is running on an EC2 Instance and used the AWS Discovery Plugin with default parameters (finding all Hazelcast members within the same region). For a more detailed description you can also check Hazelcast Guides: Deploy Hazelcast Cluster on AWS EC2.

Azure Auto Detection

To start Hazelcast on Azure Virtual Machines, you have to perform the following steps for each VM:

Attach Azure managed identity with the READ role
Install Hazelcast CLI

Then, execute the following command on each of your Virtual Machines.

hz start
...
Auto-detection selected discovery strategy: class com.hazelcast.azure.AzureDiscoveryStrategyFactory
...
Members {size:2, ver:2} [
        Member [10.11.12.1]:5701 - 3bc13eba-5d00-44cb-8ace-3e6c41bd8ead
        Member [10.11.9.5]:5701 - 6c748770-87d8-4fdf-84b1-da036f7a64bd this
]

Hazelcast members automatically discovered themselves using the Azure Discovery plugin with default parameters.

GCP Auto Detection

To start Hazelcast on GCP VM Instances, you have to perform the following steps for each instance:

Configure access to Cloud APIs (at minimum Read Only to Compute Engine API)
Install Hazelcast CLI

Then, execute the following command on each of your VM Instances.

hz start
...
Auto-detection selected discovery strategy: class com.hazelcast.gcp.GcpDiscoveryStrategyFactory
...
Members {size:2, ver:2} [
        Member [10.240.0.38]:5701 - 2aaf7b72-24d3-491c-b04f-3ca0f1aa920b this
        Member [10.240.0.40]:5701 - 222b96b5-9364-4adb-b750-711e99fc12fd
]

Hazelcast members automatically discovered themselves using the GCP Discovery plugin with default parameters.

Kubernetes Auto Detection

To start Hazelcast on Kubernetes, you need to apply the needed RBAC. Then, you can start a few Hazelcast instances and they’ll form one Hazelcast cluster.

kubectl apply -f https://raw.githubusercontent.com/hazelcast/hazelcast-kubernetes/master/rbac.yaml
kubectl run hazelcast-1 --image=hazelcast/hazelcast
kubectl run hazelcast-2 --image=hazelcast/hazelcast

That’s it, Hazelcast instances automatically discovered each other using the Kubernetes Discovery plugin with default parameters. You can check that the Hazelcast cluster was formed.

kubectl logs hazelcast-1
...
Auto-detection selected discovery strategy: class com.hazelcast.kubernetes.HazelcastKubernetesDiscoveryStrategy
...
Members {size:2, ver:2} [
        Member [10.124.1.7]:5701 - 9ed26858-8cd8-438c-a737-35308f68946e this
        Member [10.124.2.5]:5701 - d24d06f8-3ff8-4484-bdb4-759b2d459270
]

Auto Detection Under the Hood

Detecting each cloud environment requires checking some environment-specific properties. For example, to check whether Hazelcast runs on AWS EC2 Instance, we check 3 conditions and they all need to be satisfied:

Content of the file /sys/hypervisor/uuid starts with “ec2”
EC2 Instance Identity Endpoint http://169.254.169.254/latest/dynamic/instance-identity/ is reachable
IAM Role is attached to EC2 Instance

One other example is Kubernetes, for which we check the following 2 conditions:

Kubernetes-specific token file /var/run/secrets/kubernetes.io/serviceaccount/token exists
Kubernetes API is reachable via the default DNS kubernetes.default.svc

If you’re interested in details on how we detect each environment, here are the related source code references: AWS, Azure, GCP, and Kubernetes.

Auto Detection for Your Own Plugin

You can extend your own Hazelcast discovery plugin with the auto-detection functionality. Then, whenever your plugin is in the classpath, your logic checks the runtime environment and enables the plugin if needed. All you need to write is an implementation of the method DiscoveryStrategyFactory.isAutoDetectionApplicable().

public class YourDiscoveryStrategyFactory implements DiscoveryStrategyFactory {
    @Override
    public boolean isAutoDetectionApplicable() {
        // your logic to check the runtime environment
    }
}

That’s it. Hazelcast can automatically detect the runtime environment for which your plugin is designed.

Summary

Hazelcast Auto Detection feature allows you to run Hazelcast with zero configuration on your cloud environment of choice. While in the real production environment you would probably still want to provide the full Hazelcast configuration file, Auto Detection is a great way to get started!

A Hitchhiker’s Guide to Caching Patterns

Nicolas Fränkel — Tue, 08 Dec 2020 08:33:18 +0000

When your application starts slowing down, the reason is probably a bottleneck somewhere in the execution chain. Sometimes, this bottleneck is due to a bug. Sometimes, somebody didn't set up the optimal configuration. And sometimes, the process of fetching the data is the bottleneck.

One option would be to change your whole architecture. Before moving to such a drastic, and probably expensive measure, one can consider a trade-off: instead of getting remote data every time, you can store the data locally after the first read. This is the trade-off that caching offers: stale data vs. speed.

Deciding to use caching is just the first step in a long journey. The next step is to think about how your application and the cache will interact. This post focuses on your options regarding those interactions.

Cache-Aside

Cache-Aside is probably the most widespread caching pattern. With this approach, your code handles the responsibility of orchestrating the flow between the cache and the source of truth.

Regarding reads, it translates as the following:

For writes, it's even simpler:

The biggest advantage of using Cache-Aside is that anybody can read the code and understand its execution flow. Moreover, the requirements toward the cache provider are at their lowest: it just needs to be able to get and set values. That allows for pretty straightforward migrations from a cache provider to another one (e.g. Hazelcast).

The biggest issue of Cache-Aside is that your code needs to handle the inconsistency gap between the cache and the datastore. Imagine that you've successfully updated the cache but the datastore fails to update. The code needs to implement retries. Worse, during unsuccessful retries, the cache contains a value that the datastore doesn't.

Switching the logic to update the datastore first doesn't change the problem. What if the datastore updates successfully but the cache doesn't?

Read-Through

Compared to Cache-Aside, Read-Through moves the responsibility of getting the value from the datastore to the cache provider.

Read-Through implements the Separation of Concerns principle. Now, the code interacts with the cache only. It's up to the cache to manage the synchronization between itself and the datastore. It requires a more advanced cache provider than for Cache-Aside, as the former needs to provide such capability.

Hazelcast provides the MapLoader interface for this usage.

Write-Through

Similar to Read-Through but for writes, Write-Through moves the writing responsibility to the cache provider.

The main benefit of Write-Through is that the code is now free of failure handling and retry logic. Of course, it's now up to the cache to manage them.

Hazelcast provides the MapStore interface for this usage. Because in most of the cases, Write-Through also implies Read-Through, MapStore is a child-interface of MapLoader so that interactions with the datastore are co-located in the same implementation class.

Write-Behind

Write-Behind looks pretty similar to Write-Through.

I believe some of you dear readers didn't even see the difference. And if you did, you might be wondering what it does mean.

To make it clear, the difference lies in the last arrow's arrowhead: it changed from solid to line. If your UML days are past (I had to look at how to represent it), it means that the cache sends an asynchronous message to the datastore.

Up to this point, all messages exchanged between actors were synchronous: the caller needs to wait until the callee has finished processing and returned before continuing its flow. With Write-Behind, the cache sets the value to the datastore and doesn't wait for confirmation.

On the plus side, this approach speeds the whole process since the datastore is the slowest component - it sits somewhere over the network and writes to disk. On the other hand, it runs the risk of introducing inconsistencies in the cache. In Write-Through, you could retry to your heart's content until the value was set. In Write-Behind, you don't know if the set was even successful.

With Hazelcast, changing from a Write-Through approach to a Write-Behind one is just a matter of configuring the write-delay-seconds property to a value higher than 0.

Refresh-Ahead

The old saying goes that there are two hard things in computer science: naming things and cache invalidation. Cache invalidation is about planning how long an item should be stored in the cache before it expires. When it does or when the cache is still empty, you need to fetch the item from the datastore using one of the patterns above - Cache-Aside or Read-Through.

Both patterns implement a flow that involves the code, the cache, and the datastore. As mentioned above, reading from the datastore is an expensive operation: you need to first cross through the network and then request data from the datastore. What if you could prefetch the data, making it available before you even request, thus saving you from incurring the performance hit on the critical path? That's exactly what Refresh-Ahead does.

Implementations of Refresh-Ahead are cache provider-dependent. A safe bet that is agnostic to the provider is to use Hazelcast Jet. With its Change-Data-Capture capability, Jet allows to connect to any cache provider with a public API and update cached entities as soon as the datastore is updated. Here's a bird's eye view of CDC in action:

For more details on Refresh-Ahead, please check my previous post Designing an Evergreen Cache with Change-Data-Capture.

Summary

Here's a quick summary of the patterns and the context they best fit it:

Pattern	Consider	Cons
Cache-Aside	When you're limited by the capabilities of your cache provider	The application is responsible for the cache orchestration flow
Read-Through	Solid default
Write-Through	Solid default
Write-behind	When performance considerations outweigh short-term consistency	Asynchronous systems are harder to reason with
Refresh-ahead	When fetching data from the datastore impairs throughput	Additional component to develop, deploy and maintain

Announcing the Hazelcast Heroes Program

Nicolas Fränkel — Tue, 24 Nov 2020 15:17:29 +0000

Hazelcast is built on open source foundations. It’s part of our DNA, and we strive to uphold open source values in our day-to-day work. This commitment allows external contributors to make Hazelcast better so that the whole community can benefit from a collective effort.

No contribution is too small, but, indeed, some of our community members constantly go “above and beyond” to help us improve our software. For a long time, we wanted to find a way to show them how much we value their time and effort. Today, I’m very happy to make it “official” by announcing the Hazelcast Heroes program.

At the heart of the program is transparency. We offer three contribution levels, and several ways are available to achieve a level. If you meet the requirements, you automatically reap the rewards associated with the level. That’s all! No shadow committee decides on obscure and unrelated criteria those who are deserving (or not).

Without further ado, please check the program and find your place in our Hall of Heroes. We have already added our first contributors, those who trusted us, and our mission early on.

In the following months, we will reach out to people who already have met our criteria and offer them to join the program. If you’ve met the requirements and we didn’t contact you, please join our Slack channel and let us know!

Build Your Kubernetes Operator with the Right Tool

Rafał Leszko — Tue, 24 Nov 2020 12:11:01 +0000

You want to build a Kubernetes Operator for your software. Which tool to choose from? Operator SDK with Helm, Ansible, or Go? Or maybe start from scratch with Python, Java, or any other programming language? In this blog post, I discuss different approaches to writing Kubernetes Operators and list each solution’s pros and cons. All that to help you decide which tool is the right one for you!

You can find the source code for this blog post here.

Introduction

Kubernetes Operator is an application that watches a custom Kubernetes resource and performs some operations upon its changes.

This definition is very generic because the operators themselves can do a great variety of things. To make it more digestible, let’s focus on one example that we will use throughout this blog post. This example will be a Hazelcast Operator used to create and scale a Hazelcast cluster. Let’s imagine that a user wants to manage the Hazelcast cluster via a custom Kubernetes resource hazelcast.yaml.

apiVersion: hazelcast.my.domain/v1
kind: Hazelcast
metadata:
  name: hazelcast-sample
spec:
  size: 1

Upon applying this declarative configuration, a user wants to see a Hazelcast cluster with one Hazelcast member created.

$ kubectl apply -f hazelcast.yaml

Then, a user wants to be able to modify this resource configuration (i.e., change its size to 3), apply it again, and the Hazelcast cluster should resize automatically.

Hazelcast Operator is the application that watches the Hazelcast resource and interacts with Kubernetes API to create (or update) deployment with the given number of Hazelcast pod replicas.

This example is pretty simple, but in real life, a change in hazelcast.yaml could result in more complex operations. For example, cleaning data in the Hazelcast cluster, upgrading the Hazelcast version, sending metrics/logs to an external system, setting up WAN geo-replication, or creating some additional Kubernetes resources. There is no limit here; the point is that an operator observes your resource and performs any operation you want.

Operator Similarities

Since an operator is simply an application, technically, you can write it in any programming language or framework. What’s more, Kubernetes exposes a REST API, so you can really use any language to watch for Kubernetes events and to interact with the Kubernetes cluster. Nevertheless, no matter what implementation method you choose, operators have some similarities in how you build, install, and use them.

To create, install, and use an operator, you always have to:

Implement the operator logic (in your preferred language/framework)

Dockerize your operator and push it to the Docker registry

$ docker build -t <user>/hazelcast-operator . && docker push <user>/hazelcast-operator

Create CRD (Custom Resource Definition), which defines your custom resource
```
$ kubectl apply -f hazelast.crd.yaml
```
Create RBAC (Role and Role Binding) to allow an operator to interact with Kubernetes API
```
$ kubectl apply -f role.yaml
$ kubectl apply -f role_binding.yaml
```
Deploy the operator into your Kubernetes cluster
```
$ kubectl apply -f operator.yaml
```
Create your custom resource
```
$ kubectl apply -f hazelcast.yaml
```

Note that only the first point looks differently depending on the operator tool. All other steps are always exactly the same. That is why in the next sections we focus on implementing Hazelcast Operator using different techniques and for points 2-6, you can use the following source files: hazelcast.crd.yaml, role.yaml, role_binding.yaml, operator.yaml, hazelcast.yaml.

Tools for building Operators

Let’s start with the most popular tool for building operators, Operator SDK. It offers 3 approaches: Helm, Ansible, Go. Then, we’ll take a look into other operator frameworks, to close with the bare programming language implementation.

Operator SDK: Helm

Helm is a package manager for Kubernetes. It allows you to create a set of templated Kubernetes configurations, package them into a Helm chart, and then render using parameters defined in values.yaml. Helm is very simple to use because creating a Helm chart requires no more knowledge than defining standard Kubernetes configuration files.

In our example, to create a Helm-based Hazelcast Operator, we need first to create the Hazelcast Helm Chart.

$ helm create chart

Then, in chart/templates we can create deployment.yaml.

apiVersion: apps/v1
kind: Deployment
metadata:
  name: {{ include "hazelcast.fullname" . }}
spec:
  replicas: {{ .Values.size }}
  selector:
    matchLabels:
      app: hazelcast
  template:
    metadata:
      labels:
        app: hazelcast
    spec:
      containers:
        - name: hazelcast
          image: "hazelcast/hazelcast:4.1"

Our only user parameter is the cluster size and we put it into chart/values.yaml.

size: 1

Hazelcast Helm Chart is ready. Now we can generate an operator from it.

$ operator-sdk init --plugins=helm
$ operator-sdk create api --group=hazelcast --version=v1 --helm-chart=./chart

That’s it! The operator is ready. If you want to build, install, and use it, execute the common steps for all operators.

$ docker build -t leszko/hazelcast-operator . &amp;&amp; docker push leszko/hazelcast-operator
$ make install                                 # create Hazelcast CRD
$ make deploy IMG=leszko/hazelcast-operator    # create operator RBAC and install operator deployment
$ kubectl apply -f config/samples/hazelcast_v1_hazelcast.yaml # create Hazelcast resource

Note that you need to change leszko to your Docker Hub account name and make your Docker registry public.

A few comments about the “Operator SDK: Helm” approach:

Operator implementation is declarative and therefore simple; it requires no more knowledge than the standard Kubernetes configurations
If you already have a Helm chart for your software, then creating an operator requires no work at all
Your operator functionality is limited to the features available in Helm
All operator configuration files (hazelcast.crd.yaml, role.yaml, role_binding.yaml, operator.yaml hazelcast.yaml) are automatically generated, so you don’t need to maintain them separately

Operator SDK: Ansible

Ansible is a very powerful tool for IT automation. Its nature is declarative and, thanks to a number of plugins, you can write simple YAML files to perform complex DevOps tasks. Ansible has a plugin called community.kubernetes.k8s dedicated to interacting with Kubernetes API. What’s more, Operator SDK supports generating operators from Ansible roles.

To create an Ansible-based Hazelcast Operator, we need first to scaffold the Ansible operator project.

$ operator-sdk init --plugins=ansible
$ operator-sdk create api --group hazelcast --version v1 --kind Hazelcast --generate-role

Then, we can add the operator logic into roles/hazelcast/tasks/main.yml.

---
- name: start hazelcast
  community.kubernetes.k8s:
    definition:
      kind: Deployment
      apiVersion: apps/v1
      metadata:
        name: hazelcast
        namespace: '{{ ansible_operator_meta.namespace }}'
      spec:
        replicas: "{{size}}"
        selector:
          matchLabels:
            app: hazelcast
        template:
          metadata:
            labels:
              app: hazelcast
          spec:
            containers:
            - name: hazelcast
              image: "hazelcast/hazelcast:4.1"

Finally, we can add operator default parameters into roles/hazelcast/defaults/main.yml.

---
size: 1

Note that all the implementation logic looks very similar to the standard Kubernetes configuration and therefore, similar to the Helm-based operator. A significant difference is, however, that now all the configuration is interpreted by the community.kubernetes.k8s plugin, only later passed to Kubernetes API, while the Helm configuration was a direct Kubernetes configuration.

Steps to install, build, and use an Ansible-based operator are the same as for the Helm-based operator.

$ docker build -t leszko/hazelcast-operator . &amp;&amp; docker push leszko/hazelcast-operator
$ make install                                 # create Hazelcast CRD
$ make deploy IMG=leszko/hazelcast-operator    # create operator RBAC and install operator deployment
$ kubectl apply -f config/samples/hazelcast_v1_hazelcast.yaml # create Hazelcast resource

A few comments about the “Operator SDK: Ansible” approach:

Ansible allows implementing operator in a declarative form which is concise and human-readable
Ansible operator is similar to the pure Kubernetes configuration but executed via the community.kubernetes.k8s plugin
Ansible is a very powerful tool and it lets you express almost any logic you may want
Similar to Helm-based operator, all configuration files (hazelcast.crd.yaml, role.yaml, role_binding.yaml, operator.yaml hazelcast.yaml) are automatically generated, so no need for any additional maintenance

Operator SDK: Go

Go is a general-purpose programming language, so you can technically write any operator logic you could ever imagine. What’s more, the Kubernetes environment itself is written in Go, so the Kubernetes client library (interacting with Kubernetes API) is second to none. Operator SDK (with embedded Kubebuilder) supports implementing operators in Go, so you get a lot of scaffolding and code generation for free.

To create a Go-based Hazelcast Operator, we need first to execute a few commands to scaffold the project.

$ operator-sdk init --repo=github.com/leszko/hazelcast-operator
$ operator-sdk create api --version v1 --group=hazelcast --kind Hazelcast --resource=true --controller=true

Then, we are ready to implement the operator logic inside the function Reconcile() of the file controllers/hazelcast_controller.go.

// +kubebuilder:rbac:groups=hazelcast.my.domain,resources=hazelcasts,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=hazelcast.my.domain,resources=hazelcasts/status,verbs=get;update;patch
// +kubebuilder:rbac:groups=apps,resources=deployments,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=core,resources=pods,verbs=get;list;watch

func (r *HazelcastReconciler) Reconcile(req ctrl.Request) (ctrl.Result, error) {
    ctx := context.Background()
    log := r.Log.WithValues("hazelcast", req.NamespacedName)

    // Fetch the Hazelcast instance
    hazelcast := &amp;hazelcastv1.Hazelcast{}
    err := r.Get(ctx, req.NamespacedName, hazelcast)
    if err != nil {
        if errors.IsNotFound(err) {
            log.Info("Hazelcast resource not found. Ignoring since object must be deleted")
            return ctrl.Result{}, nil
        }
        log.Error(err, "Failed to get Hazelcast")
        return ctrl.Result{}, err
    }

    // Check if the deployment already exists, if not create a new one
    found := &amp;appsv1.Deployment{}
    err = r.Get(ctx, types.NamespacedName{Name: hazelcast.Name, Namespace: hazelcast.Namespace}, found)
    if err != nil &amp;&amp; errors.IsNotFound(err) {
        // Define a new deployment
        dep := r.deploymentForHazelcast(hazelcast)
        log.Info("Creating a new Deployment", "Deployment.Namespace", dep.Namespace, "Deployment.Name", dep.Name)
        err = r.Create(ctx, dep)
        if err != nil {
            log.Error(err, "Failed to create new Deployment", "Deployment.Namespace", dep.Namespace, "Deployment.Name", dep.Name)
            return ctrl.Result{}, err
        }
        return ctrl.Result{Requeue: true}, nil
    } else if err != nil {
        log.Error(err, "Failed to get Deployment")
        return ctrl.Result{}, err
    }

    // Ensure the deployment size is the same as the spec
    size := hazelcast.Spec.Size
    if *found.Spec.Replicas != size {
        found.Spec.Replicas = &amp;size
        err = r.Update(ctx, found)
        if err != nil {
            log.Error(err, "Failed to update Deployment", "Deployment.Namespace", found.Namespace, "Deployment.Name", found.Name)
            return ctrl.Result{}, err
        }
        return ctrl.Result{Requeue: true}, nil
    }

    // Update the Hazelcast status with the pod names
    // List the pods for this hazelcast's deployment
    podList := &amp;corev1.PodList{}
    listOpts := []client.ListOption{
        client.InNamespace(hazelcast.Namespace),
        client.MatchingLabels(labelsForHazelcast(hazelcast.Name)),
    }
    if err = r.List(ctx, podList, listOpts...); err != nil {
        log.Error(err, "Failed to list pods", "Hazelcast.Namespace", hazelcast.Namespace, "Hazelcast.Name", hazelcast.Name)
        return ctrl.Result{}, err
    }
    podNames := getPodNames(podList.Items)

    // Update status.Nodes if needed
    if !reflect.DeepEqual(podNames, hazelcast.Status.Nodes) {
        hazelcast.Status.Nodes = podNames
        err := r.Status().Update(ctx, hazelcast)
        if err != nil {
            log.Error(err, "Failed to update Hazelcast status")
            return ctrl.Result{}, err
        }
    }

    return ctrl.Result{}, nil
}

// deploymentForHazelcast returns a hazelcast Deployment object
func (r *HazelcastReconciler) deploymentForHazelcast(m *hazelcastv1.Hazelcast) *appsv1.Deployment {
    ls := labelsForHazelcast(m.Name)
    replicas := m.Spec.Size

    dep := &amp;appsv1.Deployment{
        ObjectMeta: metav1.ObjectMeta{
            Name:      m.Name,
            Namespace: m.Namespace,
        },
        Spec: appsv1.DeploymentSpec{
            Replicas: &amp;replicas,
            Selector: &amp;metav1.LabelSelector{
                MatchLabels: ls,
            },
            Template: corev1.PodTemplateSpec{
                ObjectMeta: metav1.ObjectMeta{
                    Labels: ls,
                },
                Spec: corev1.PodSpec{
                    Containers: []corev1.Container{{
                        Image: "hazelcast/hazelcast:4.1",
                        Name:  "hazelcast",
                    }},
                },
            },
        },
    }
    // Set Hazelcast instance as the owner and controller
    ctrl.SetControllerReference(m, dep, r.Scheme)
    return dep
}

// labelsForHazelcast returns the labels for selecting the resources
// belonging to the given hazelcast CR name.
func labelsForHazelcast(name string) map[string]string {
    return map[string]string{"app": "hazelcast", "hazelcast_cr": name}
}

// getPodNames returns the pod names of the array of pods passed in
func getPodNames(pods []corev1.Pod) []string {
    var podNames []string
    for _, pod := range pods {
        podNames = append(podNames, pod.Name)
    }
    return podNames
}</pre>
<p>We also need to add the size field to the Hazelcast resource structure in <a href="https://github.com/leszko/build-your-operator/blob/main/operator-sdk-go/api/v1/hazelcast_types.go">api/v1/hazelcast_types.go</a>.</p>
<pre>type HazelcastSpec struct {
    Size int32 `json:"size,omitempty"`
}

You can already see that we had to write way more code and that this code is much more complex than the previous Operator SDK solutions. That’s all because we came from the declarative Kubernetes configurations to the imperative programming language. That means that now it’s not enough to change the size in the configuration, but we need to provide the code flow (with the proper error handling). That’s definitely more difficult! On the other hand, programming language gives you the flexibility to program anything you want. The declarative Kubernetes configuration no longer limits you and your operator can perform any logic you could ever imagine.

One thing to note is the list of comments above the Reconcile() function. They are used by Operator SDK to generate role.yaml for the operator.

Steps to install, build, and use a Go-based operator are the same as for any other Operator SDK operator.

$ docker build -t leszko/hazelcast-operator . &amp;&amp; docker push leszko/hazelcast-operator
$ make install                                 # create Hazelcast CRD
$ make deploy IMG=leszko/hazelcast-operator    # create operator RBAC and install operator deployment
$ kubectl apply -f config/samples/hazelcast_v1_hazelcast.yaml # create Hazelcast resource

A few comments about the “Operator SDK: Go” approach:

You implement your operator in an imperative code, which requires more work and caution
Go language is well integrated with Kubernetes
Writing an operator in the real programming language as Go means no limits on the functionality you want to implement
Operator SDK helps to scaffold the Go operator project as well as generating boilerplate configuration files (hazelcast.crd.yaml, role.yaml, role_binding.yaml, operator.yaml, hazelcast.yaml)

Operator Framework: KOPF

Operator SDK is the most popular tool for creating operators, but it’s not the only one. You can find some other interesting solutions for most programming languages, for example, Java Operator SDK or Kubernetes Operator Pythonic Framework (KOPF). The former one gained some traction because its implementation was started inside the Zalando company.

To start creating the KOPF-based Hazelcast Operator, we first need to manually prepare all the boilerplate files: Dockerfile and hazelcast.crd.yaml, role.yaml, role_binding.yaml, operator.yaml, and hazelcast.yaml. Then, we’re ready to write the operator logic in the operator.py file.

import kopf
import pykube
import yaml

@kopf.on.create('hazelcast.my.domain', 'v1', 'hazelcasts')
def create_fn(spec, **kwargs):
    doc = create_deployment(spec)
    kopf.adopt(doc)

    api = pykube.HTTPClient(pykube.KubeConfig.from_env())
    deployment = pykube.Deployment(api, doc)
    deployment.create()

    api.session.close()

    return {'children': [deployment.metadata['uid']]}


@kopf.on.update('hazelcast.my.domain', 'v1', 'hazelcasts')
def update_fn(spec, **kwargs):
    api = pykube.HTTPClient(pykube.KubeConfig.from_env())
    deployment = pykube.Deployment.objects(api).get(name="hazelcast")
    deployment.replicas = spec.get('size', 1)
    deployment.update()

    api.session.close()

    return {'children': [deployment.metadata['uid']]}

def create_deployment(spec):
    return yaml.safe_load(f"""
        apiVersion: apps/v1
        kind: Deployment
        metadata:
          name: hazelcast
        spec:
          replicas: {spec.get('size', 1)}
          selector:
            matchLabels:
              app: hazelcast
          template:
            metadata:
              labels:
                app: hazelcast
            spec:
              containers:
                - name: hazelcast
                  image: "hazelcast/hazelcast:4.1"
    """)

Python itself is quite a concise language and, thanks to using Python decorators, the code looks short, clean, and tidy. However, similar to Go-based implementation, you need to cover each operation (create and update) separately because we write the imperative code. One difference compared to Go is that the Python Kubernetes client is not as well integrated with Kubernetes as the Go Kubernetes client. Python uses YAML Kubernetes configurations and manipulates them, while Go operates on Kubernetes structures.

Steps to install, build, and use a KOPF-based operator are similar to what we saw before.

$ docker build -t leszko/hazelcast-operator:kopf . &amp;&amp; docker push leszko/hazelcast-operator:kopf
$ kubectl apply -f hazelcast.crd.yaml # create Hazelcast CRD
$ kubectl apply -f role.yaml          # create operator RBAC
$ kubectl apply -f role_binding.yaml  # create operator RBAC
$ kubectl apply -f operator.yaml      # install operator
$ kubectl apply -f hazelcast.yaml     # create Hazelcast resource

A few comments about the “Operator Framework” approach:

Operator Frameworks for specific languages are less developed and popular than Operator SDK
While Operator SDK provides project scaffolding and boilerplate code generation, Operator Frameworks usually leave this work to a developer
Kubernetes clients for Python, Java, or other languages are always slightly worse choice than the native Go Kubernetes client
Programming in any general-purpose language like Python or Java means that there is no limit on the functionary your operator provides

Bare Programming Language: Java

Operators are nothing more than dockerized applications, so technically you can write them in any programming language. Such a solution means, however, that you’re on your own. Nothing helps you in generating all the boilerplate code or configurations. On the other hand, you can choose the language you already use for other projects in your enterprise, decreasing the learning curve. A blog post Writing a Kubernetes Operator in Java describes how to implement an operator with Java, using Quarkus to increase the performance by building Docker native images. Let’s take a similar approach and create a Java-based Hazelcast Operator.

We need first to manually prepare all the boilerplate files: hazelcast.crd.yaml, role.yaml, role_binding.yaml, operator.yaml, and hazelcast.yaml. We can then scaffold a Quarkus project using quarkus-maven-plugin or just clone the source repository for this blog post. Finally, we can implement the operator logic in a few classes.

public class ClientProvider {

    @Produces
    @Singleton
    @Named("namespace")
    private String findNamespace() throws IOException {
        return new String(Files.readAllBytes(Paths.get("/var/run/secrets/kubernetes.io/serviceaccount/namespace")));
    }

    @Produces
    @Singleton
    KubernetesClient newClient(@Named("namespace") String namespace) {
        return new DefaultKubernetesClient().inNamespace(namespace);
    }

    @Produces
    @Singleton
    NonNamespaceOperation&lt;HazelcastResource, HazelcastResourceList, HazelcastResourceDoneable, Resource&lt;HazelcastResource, HazelcastResourceDoneable&gt;&gt; makeCustomResourceClient(
            KubernetesClient defaultClient, @Named("namespace") String namespace) {

        KubernetesDeserializer.registerCustomKind("hazelcast.my.domain/v1", "Hazelcast", HazelcastResource.class);

        CustomResourceDefinition crd = defaultClient
                .customResourceDefinitions()
                .list()
                .getItems()
                .stream()
                .filter(d -&gt; "hazelcasts.hazelcast.my.domain".equals(d.getMetadata().getName()))
                .findAny()
                .orElseThrow(
                        () -&gt; new RuntimeException(
                                "Deployment error: Custom resource definition \"hazelcasts.hazelcast.my.domain\" not found."));

        return defaultClient
                .customResources(crd, HazelcastResource.class, HazelcastResourceList.class, HazelcastResourceDoneable.class)
                .inNamespace(namespace);
    }
}

@ApplicationScoped
public class DeploymentInstaller {

    @Inject
    private KubernetesClient client;

    @Inject
    private HazelcastResourceCache cache;

    void onStartup(@Observes StartupEvent _ev) {
        new Thread(this::runWatch).start();
    }

    private void runWatch() {
        cache.listThenWatch(this::handleEvent);
    }

    private void handleEvent(Watcher.Action action, String uid) {
        try {
            HazelcastResource resource = cache.get(uid);
            if (resource == null) {
                return;
            }

            Predicate ownerRefMatches = deployments -&gt; deployments.getMetadata().getOwnerReferences().stream()
                    .anyMatch(ownerReference -&gt; ownerReference.getUid().equals(uid));


            List hazelcastDeployments = client.apps().deployments().list().getItems().stream()
                    .filter(ownerRefMatches)
                    .collect(toList());

            if (hazelcastDeployments.isEmpty()) {
                client.apps().deployments().create(newDeployment(resource));
            } else {
                for (Deployment deployment : hazelcastDeployments) {
                    setSize(deployment, resource);
                    client.apps().deployments().createOrReplace(deployment);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
            System.exit(-1);
        }
    }

    private Deployment newDeployment(HazelcastResource resource) {
        Deployment deployment = client.apps().deployments().load(getClass().getResourceAsStream("/deployment.yaml")).get();
        setSize(deployment, resource);
        deployment.getMetadata().getOwnerReferences().get(0).setUid(resource.getMetadata().getUid());
        deployment.getMetadata().getOwnerReferences().get(0).setName(resource.getMetadata().getName());
        return deployment;
    }

    private void setSize(Deployment deployment, HazelcastResource resource) {
        deployment.getSpec().setReplicas(resource.getSpec().getSize());
    }
}

@ApplicationScoped
public class HazelcastResourceCache {

    private final Map&lt;String, HazelcastResource&gt; cache = new ConcurrentHashMap&lt;&gt;();

    @Inject
    private NonNamespaceOperation&lt;HazelcastResource, HazelcastResourceList, HazelcastResourceDoneable, Resource&lt;HazelcastResource, HazelcastResourceDoneable&gt;&gt; crClient;

    private Executor executor = Executors.newSingleThreadExecutor();

    public HazelcastResource get(String uid) {
        return cache.get(uid);
    }

    public void listThenWatch(BiConsumer&lt;Watcher.Action, String&gt; callback) {

        try {
            // list
            crClient
                    .list()
                    .getItems()
                    .forEach(resource -&gt; {
                                cache.put(resource.getMetadata().getUid(), resource);
                                String uid = resource.getMetadata().getUid();
                                executor.execute(() -&gt; callback.accept(Watcher.Action.ADDED, uid));
                            }
                    );

            // watch
            crClient.watch(new Watcher() {
                @Override
                public void eventReceived(Action action, HazelcastResource resource) {
                    try {
                        String uid = resource.getMetadata().getUid();
                        if (cache.containsKey(uid)) {
                            int knownResourceVersion = Integer.parseInt(cache.get(uid).getMetadata().getResourceVersion());
                            int receivedResourceVersion = Integer.parseInt(resource.getMetadata().getResourceVersion());
                            if (knownResourceVersion &gt; receivedResourceVersion) {
                                return;
                            }
                        }
                        System.out.println("received " + action + " for resource " + resource);
                        if (action == Action.ADDED || action == Action.MODIFIED) {
                            cache.put(uid, resource);
                        } else if (action == Action.DELETED) {
                            cache.remove(uid);
                        } else {
                            System.err.println("Received unexpected " + action + " event for " + resource);
                            System.exit(-1);
                        }
                        executor.execute(() -&gt; callback.accept(action, uid));
                    } catch (Exception e) {
                        e.printStackTrace();
                        System.exit(-1);
                    }
                }

                @Override
                public void onClose(KubernetesClientException cause) {
                    cause.printStackTrace();
                    System.exit(-1);
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
            System.exit(-1);
        }
    }
}

Additionally, we need to add the Kubernetes configuration file src/main/resources/deployment.yaml, used in the Java code.

apiVersion: apps/v1
kind: Deployment
metadata:
  name: hazelcast
  ownerReferences:
    - apiVersion: apps/v1
      kind: Hazelcast
      name: placeholder
      uid: placeholder
spec:
  replicas: 1
  selector:
    matchLabels:
      app: hazelcast
  template:
    metadata:
      labels:
        app: hazelcast
    spec:
      containers:
        - name: hazelcast
          image: "hazelcast/hazelcast:4.1"

Apart from the code above, we need to add additional Java boilerplate classes: HazelcastResource, HazelcastResourceDoneable, HazelcastResourceList, HazelcastResourceSpec. Yes… the combination of using Java and not using any Operator Framework must result in a lot of code. A lot of code to write and a lot of code to maintain. Java is verbose; Java Kubernetes client is verbose. That’s it. For details of the code above, I recommend reading Writing a Kubernetes Operator in Java. You can also check there the tweaks you need to make to build the Docker native image.

We can build the application in two ways: Java Docker image or Docker native image. The second approach is better for the performance but requires a few tweaks in code, so for the purpose of this blog post, let’s just build a standard Docker image (and then install and use it),

$ mvn package
$ docker build -f src/main/docker/Dockerfile.jvm -t leszko/hazelcast-operator:java . &amp;&amp; docker push leszko/hazelcast-operator:java
$ kubectl apply -f hazelcast.crd.yaml # create Hazelcast CRD
$ kubectl apply -f role.yaml          # create operator RBAC
$ kubectl apply -f role_binding.yaml  # create operator RBAC
$ kubectl apply -f operator.yaml      # install operator
$ kubectl apply -f hazelcast.yaml     # create Hazelcast resource

A few comments about the “Bare Programming Language” approach:

Creating an operator from scratch means writing more code
There is no limit on the logic you want to deliver
Before starting to write an operator in your preferred language, you can check if it provides a good and mature Kubernetes client library
The only good reason to write an operator from scratch is using a single programming language inside your project/organization/enterprise

Summary

You can look at the code snippets above and decide which operator implementation is the right one for you. However, that’s just a part of the story. Let me give you an example. I’m a Java developer, so for me using pure Java is the simplest approach. Still, I would never choose Java for writing an operator. Why? Most developers do not write operators in Java. So, I’d be alone in it! Alone with bugs, alone with new features, alone with my questions on StackOverflow. And programming is a collaborative work!

Then, what operator tool do others use? Let’s look at the data from OperatorHub.io.

Go-based operators are by far the most popular. You may find the data slightly biased because the operators published at OperatorHub are only those operators that are built and distributed for others, so you won’t find any internal operators there. But still, if you decide to develop your operator in Go, you’re in good company!

So, which operator implementation is the right one for you? The choice is yours, but let me give you some hints.

Hint 1: If you already have a Helm chart for your software and you don’t need any complex capability levels => Operator SDK: Helm
Hint 2: If you want to create your operator quickly and you don’t need any complex capability levels => Operator SDK: Helm
Hint 3: If you want complex features or/and be flexible about any future implementations => Operator SDK: Go
Hint 4: If you want to keep a single programming language in your organization
- If a popular Operator Framework exists for your language or/and you want to contribute to it => Operator Framework
- If no popular Operator Framework exists for your programming language => Bare Programming Language
Hint 5: If none of the above => Operator SDK: Go

Our Journey to a High-Performance Node.js Library

Andrei Pechkurov — Thu, 19 Nov 2020 06:47:15 +0000

As you may already know, the Hazelcast In-Memory Data Grid (IMDG) ecosystem includes a variety of clients for different languages and runtimes, which includes Node.js client library as a part of that list.

You can use Hazelcast clients in various cases, including, but not limited to the following:

Building a multi-layer cache for your applications with IMap, a distributed, replicated key-value store, and its NearCache.
Enabling pub-sub communication between application instances.
Dealing with high load for views or likes events by using a conflict-free replicated counter.
Preventing races when accessing 3rd-party services by using FencedLock and other distributed concurrency primitives available in Hazelcast CP Subsystem (powered by Raft consensus algorithm).

High performance and low latency for data access have always been a key feature of Hazelcast. So, it’s not surprising that we put a lot of time and effort into optimizing both server-side and client libraries.

Our Node.js library went through numerous performance analysis and optimization runs over the course of several releases, and we think it’s worth telling you the story and sharing the gathered experience. If you develop a library or an application for Node.js and performance is something you care about, you may find this blog post valuable.

TL;DR

Performance analysis is not a one-time action but rather a (sometimes tiring) process.
Node.js core and the ecosystem includes useful tools, like the built-in profiler, to help you with the analysis.
Be prepared for the fact that you will have to throw many (if not most) of your experiments into the trash as part of the optimization process.
While “high-performance library” title may sound too loud, we do our best to deserve it for Node.js and all the other Hazelcast client libraries.

We’re going to start this story in spring 2019, in the times of 0.10.0 version of the Node.js client. Back then, the library was more or less feature complete, but there was little understanding of its performance. Obviously, it was necessary to analyze the performance before the first non-0.x release of the client and that’s where this story starts.

Benchmarks

It’s not a big secret that benchmarking is tricky. Even VMs themselves may introduce noticeable variation in results and even fail to reach a steady performance state. Add Node.js, library, and benchmark code on top of that and the goal of reliable benchmarking will get even harder. Any performance analysis has to rely on inputs provided by some kind of benchmark. Luckily, version 0.10.0 of the library included a simple benchmark used in early development phases. That benchmark had some limitations which needed to be resolved before going any further.

The existing benchmark supported only a single scenario with randomly chosen operations. There is nothing wrong with having a random-based scenario in the benchmark suite, but only when more narrow scenarios are present in the suite. In the case of a client library, that would be “read-heavy” and “write-heavy” scenarios. The first assumes sending lots of read operations, thus moving the hot path to the I/O read-from-socket code and further data deserialization. You may have already guessed that the second scenario involves lots of writes and moves write-to-socket and serialization code to the hot path. So, we added these additional scenarios.

Another noticeable addition to scenarios was support for the payload size option. Variation in payload size is important when running benchmarks, as it helps with finding potential bottlenecks in the serialization code. Using different payload types is also valuable, but for a start, we decided to deal with strings only. String type is used for storing JSON data on the Hazelcast cluster, so our choice had a nice side-effect of testing a significant part of the hot path for JSON payload type (i.e., for plain JavaScript objects).

The second problem was self-throttling of the benchmark. Simply put, the benchmark itself was acting as a bottleneck hiding real bottleneck issues present in the client library. Each next operation run by the benchmark was scheduled with the setImmediate() function without any concurrency limit for the sent operations. Apart from becoming a bottleneck, this approach also created a significant level of noise (sometimes it’s called “jitter”) in the benchmark results. Even worse, such logic puts the benchmark very far from real-world Node.js applications.

That’s why we improved the benchmark by enforcing the given concurrency limit. The end behavior of our benchmark runner is close to the popular p-limit package and can be visualized as the following diagram:

The diagram shows how operations are executed when the concurrency limit is set to 3 and the total count of operations to be run is 7. As a result, the load put on both the client and the server-side instances is evenly distributed, which helps to minimize the jitter.

Finally, we added a warm-up phase into the benchmark to give both client and server VMs some time to reach a steady state.

Now, with our new shiny benchmark, we were ready to start the actual analysis.

Here Come the Bottlenecks

The very first benchmark run showed the following results in scenarios based on IMap’s get() (“read-heavy”) and set() (“write-heavy”) operations.

Scenario	get() 3B	get() 1KB	get() 100KB	set() 3B	set() 1KB	set() 100KB
Throughput (ops/sec)	90,933	23,591	105	76,011	44,324	1,558

Each result here stands for an average throughput calculated over a number of benchmark runs. Result variation, median and outliers are omitted for the sake of brevity, but they were also considered when comparing results.

Data sizes (3B, 1KB, and 100KB) in the table stand for the value size. Of course, absolute numbers are not important here, as we didn’t yet have a baseline. Still, the results for the smallest value size look more or less solid and, if we would only run these benchmarks, we could stop the analysis, give the library a green light for the first major release, and arrange the release party. But results for larger values are much more disturbing. They scale down almost linearly with the growth of the value size, which doesn’t look good. This gave us a clue that there was a bottleneck somewhere on the hot path, presumably in the serialization code. Further analysis was required.

Node.js is quite mature and there are a number of tools in the ecosystem to help you with finding bottlenecks. The first one is the V8’s sampling profiler exposed by Node.js core. It collects information about call stacks in your application with a constant time interval and stores it in an intermediate profile file. Then it allows you to prepare a text report based on the profile. The core logic is simple: the more samples contain a function on the top of the call stack, the more time was spent in the function when profiling. Thus, potential bottlenecks are usually found among the most “heavy” functions.

Profiler reports are helpful in many situations, but sometimes you may want to start the analysis with visual information. Fortunately, flame graphs are there to help. There are a number of ways to collect flame graphs for Node.js applications, but we were more than fine with 0x library.

Here is a screenshot of the flame graph collected for the set() 3B scenario.

This screenshot is static, while 0x produces an interactive web page allowing you to zoom and filter through the contents of the flame graph. In this particular case, it took us some time to iterate over so-called “platos” in search of suspicious calls. Finally, we found a good candidate highlighted in the next picture.

It appeared that the library was doing a lot of unnecessary allocations for Buffer objects. Buffers are low-level objects based on V8’s ArrayBuffer class, which represents contiguous arrays of binary data. The actual data is stored off-heap (there are some exceptions to this rule, but they are not relevant for our case), so allocating a Buffer may be a relatively expensive operation.

As a simple fix, we tried to get rid of certain Buffer allocations happening in the library by doing those allocations in a greedy manner. With this change, the benchmark showed us the following.

	get() 3B	get() 1KB	get() 100KB	set() 3B	set() 1KB	set() 100KB
v0.10.0	90,933	23,591	105	76,011	44,324	1,558
Candidate	104,854	24,929	109	95,165	52,809	1,581
	+15%	+5%	+3%	+25%	+19%	+1%

The improvement was noticeable for smaller payloads, but the scalability issue was still there. While the fix was very simple, if not primitive, the very first bottleneck was found. The fix was good enough as the initial optimization and further improvements were put into the backlog for future versions of the library.

The next step was to analyze so-called “read-heavy” scenarios. After a series of profiler runs and a thoughtful analysis, we found a suspicious call. The call is highlighted on the following screenshot for get() 100KB flame graph.

The ObjectDataInput.readUtf() method appeared to be executed on a significant percentage of collected profiler samples, so we started looking into that. The method was responsible for string deserialization (i.e., creating a string from the binary data) and looked more or less like the following TypeScript code.

private readUTF(pos?: number): string {
  const len = this.readInt(pos);
  // ...
  for (let i = 0; i < len; i++) {
    let charCode: number;
    leadingByte = this.readByte(readingIndex) & MASK_1BYTE;
    readingIndex = this.addOrUndefined(readingIndex, 1);
    const b = leadingByte & 0xFF;
    switch (b >> 4) {
      // ...
    }
    result += String.fromCharCode(charCode);
  }
  return result;
}

In general, the method was similar to what we had in the Hazelcast Java client. It was reading UTF-8 chars one by one and concatenating the result string. That looked like a suboptimal code, considering that Node.js provides the buf.toString() method as a part of the standard library. To compare these two implementations, we wrote simple microbenchmarks for both string deserialization and serialization. Here is a trimmed result for the serialization microbenchmark.

As it is clearly seen here, the standard API is significantly (around x6) faster than our custom implementation when it comes to ASCII strings (which are a frequent case in user applications). Results for deserialization and other scenarios look similar with the respect to the string size correlation. That was the exact reason for the scalability issue.

The standard library is significantly faster in the ASCII string case, as V8 is smart enough to detect the case and go over the fast path where it simply copies string contents instead of decoding/encoding individual chars. For those of you who are curious about the corresponding V8 source code, here is the place responsible for the buf.toString()’s fast path.

Anyhow, before making the final verdict, it was necessary to confirm the hypothesis with a proper experiment. To do so, we implemented a fix and compared it with the baseline (v0.10.0).

	get() 3B	get() 1KB	get() 100KB	set() 3B	set() 1KB	set() 100KB
v0.10.0	90,933	23,591	105	76,011	44,324	1,558
Candidate	122,458	104,090	7,052	110,083	73,618	8,428
	+34%	+341%	+6,616%	+45%	+66%	+440%

Bingo! Lesson learned: always bet on the standard library. Even if it’s slower today, things may change dramatically in the future releases.

As a result of this short (~1.5 weeks) initial analysis, Hazelcast Node.js client v3.12 was released with both of the discussed performance improvements.

Now, when there is an understanding of our usual process, let’s speed up the narration and briefly describe optimizations shipped in later versions of the library.

Automated Pipelining

Protocol pipelining is a well-known technique used to improve the performance of blocking APIs. On the user level, it usually implies an explicit batching API, which is only applicable to a number of use cases, like ETL pipelines.

Obviously, the same approach can be applied to Node.js with its non-blocking APIs. But we wanted to apply the technique in an implicit fashion so that most applications would benefit from the new optimization. We ended up with the feature called automated pipelining. It can be illustrated with the following diagram.

The main idea is to accumulate outbound messages based on the event loop lifecycle instead of writing them into a TCP socket immediately when the user starts an operation. The messages are scheduled to be concatenated into a single Buffer (with a configured size threshold) and only then are written into the socket. This way we benefit from batch writes without having to ask the user to deal with an explicit pipelining API.

Another important aspect here is that the client keeps one persistent connection per cluster member (note: we’re talking of smart client mode). Consequently, network communication over each connection is intensive enough to make the described batching logic valuable in terms of throughput.

Hazelcast Java client implements something close to this optimization by concatenating messages before writing them into the socket. A similar approach is used in other Node.js libraries, like DataStax Node.js driver for Apache Cassandra.

Benchmark measurements for automated pipelining showed 24-35% throughput improvement in read and write scenarios. The only drawback was a certain degradation (~23%) in scenarios with large message writes (100KB), which is expected considering the nature of the optimization. As real-world applications read data more frequently than write it, it was decided to enable automated pipelining by default and allow users to disable it via the client configuration.

Later on, we have improved automated pipelining by optimizing the code, which was manipulating the write queue. The main improvement came from reusing the outbound Buffer instead of allocating a new one on each write. Apart from this, we also were able to get rid of the remaining unnecessary Buffer allocations that we had in the library. As a result, we got around 8-10% throughput improvement. This latest version of automated pipelining may be found in the 4.0 release of the client.

Boomerang Backups

As you may guess, it’s not all about Node.js specific optimizations. Periodically, all Hazelcast clients get common optimizations. Client backup acknowledgments (a.k.a. boomerang backups) are a recent example of this process.

Previously, the client was waiting for the sync backups to complete on the member. This was causing 4 network hops to complete a client operation with sync backup. Since sync backup configuration is our out-of-the-box experience, boomerang backups optimization was introduced. The following diagram illustrates the change in terms of client-to-cluster communication.

As it may be seen above, boomerang backups decrease network hops to 3. With this change, we saw up to 30% throughput improvement in our tests. This optimization was shipped in client v4.0.

Migration to Native Promises

Everyone knows that callbacks lost the battle and most Node.js applications are written with promises. That’s why Hazelcast Node.js client had a Promise-based API from the day one. In older versions, it was using the bluebird Promise library for performance reasons. But since then, V8’s native Promise implementation got much faster and we decided to give native promises a try.

Benchmark measurements showed no performance regression after the migration, so the switch was shipped in v4.0. As a nice side effect of this change, we got an out-of-the-box integration with async_hooks module.

Other Optimizations

Expectedly, there were a bunch of smaller optimizations done on the way. Say, to reduce the amount of litter generated on the hot path we switched from new Date() calls to Date.now(). Another example is the default serializer implementation for Buffer objects. It allows users to deal with Buffers instead of plain arrays of numbers. Not saying that the internal code responsible for manipulations with Buffers also improved a lot. It’s hard to notice an effect of individual optimization here, but they’re certainly worth it.

A Self-Check

Before the wrap-up, let’s try to look at what we achieved in about one year. To do so, we’re going to run a couple of benchmarks for versions 0.10.0 (our baseline) and 4.0 (the latest one).

For the sake of brevity we’re going to compare IMap.set() and get() operations for 1KB ASCII values. Hopefully, the payload is close enough to what one may see on average in Node.js applications. Here is how the result looks like.

In the above chart, we see almost x3 throughput improvement in both operations. The value of all implemented optimizations should be obvious now.

What’s Next?

There are multiple things we want to give a try in both the library and the tooling. For instance, we’re experimenting with the onread option available in the net.Socket class. This option allows one to reuse Buffer when reading from the socket. Unfortunately, tls module used by the client for encrypted communication lacks the counterpart option, so recently we contributed to the Node.js core to improve things.

Our benchmarking approach also needs some improvements. First of all, we want to start considering operation latency by collecting latency data into an HDR histogram throughout benchmark execution. Another nice addition would be integration with Hazelcast Simulator, our distributed benchmarking framework. Finally, support for more data structures and payload types won’t hurt.

Lessons Learned

Yes, we know that the “high-performance library” title may be too loud, but we do our best to deserve it. For us, as open-source library maintainers, performance analysis is a process that requires constant attention. Necessary routing actions, like pre-release performance analysis, may be tiring. We had to throw many (if not most) of our experiments into the trash can. But in the end, performance is something we aim to deliver in all of our client libraries.

Hazelcast Node.js Client 4.0 is Released

Andrei Pechkurov — Fri, 02 Oct 2020 07:10:20 +0000

Hazelcast Node.js client 4.0 is now available! Let’s see what are the main changes in this new release.

Hazelcast Client Protocol 2.0

Node.js client now uses Hazelcast Open Binary Client Protocol 2.0, which has a number of enhancements and serialization improvements when compared with 1.x. For the end-user, it means that the client now supports IMDG 4.0+. Also, note that you cannot use a 4.0 client with IMDG 3.x members.

Ownerless Client

In Hazelcast 3.x, clients were implicitly assigned to an owner member responsible for cleaning up their resources after they leave the cluster. Ownership information had to be replicated to the whole cluster when a client joined the cluster. The “owner member” concept is now removed and Node.js client 4.0 acts as an ownerless client, which is a simpler solution for the problem allowing to remove the extra step.

Configuration Redesign and API Cleanup

Programmatic configuration in client 4.0 has become simpler and does not require boilerplate code anymore. The configuration itself is now represented with a plain JavaScript object.

Programmatic configuration (old way):

const { Client, Config } = require('hazelcast-client');

// Create a configuration object
const clientConfig = new Config.ClientConfig();

// Customize the client configuration
clientConfig.clusterName = 'cluster-name';
clientConfig.networkConfig.addresses.push('10.90.0.2:5701');
clientConfig.networkConfig.addresses.push('10.90.0.3:5701');
clientConfig.listeners.addLifecycleListener(function (state) {
    console.log('Lifecycle Event >>> ' + state);
});

// Initialize the client with the given configuration
const client = await Client.newHazelcastClient(clientConfig);

Programmatic configuration (new way):

// No need to require Config anymore
const { Client } = require('hazelcast-client');

// Initialize the client with the configuration object (POJO)
const client = await Client.newHazelcastClient({
    clusterName: 'cluster-name',
    network: {
        clusterMembers: [
            '10.90.0.2:5701',
            '10.90.0.3:5701'
        ]
    },
    lifecycleListeners: [
        (state) => {
            console.log('Lifecycle Event >>> ' + state);
        }
    ]
});

The “shape” of the configuration is kept close to the old declarative configuration API and to the Java client’s YAML/XML configuration. So, the user experience is the same across other Hazelcast clients, but it is also native to JavaScript and Node.js runtime.

The old declarative configuration API was removed as it does not make a lot of sense now, considering these changes.

The 4.0 release also brings a number of changes aimed to make the API more idiomatic for JavaScript and familiar to Node.js developers.

CP Subsystem Support

In Hazelcast 4.0, concurrent primitives moved to CP Subsystem. CP Subsystem contains new implementations of Hazelcast’s concurrency APIs on top of the Raft consensus algorithm. As the name of the module implies, these implementations are CP with respect to the CAP principle and they live alongside the AP data structures in the same Hazelcast IMDG cluster. They maintain linearizability in all cases, including client and server failures, network partitions, and prevent split-brain situations.

Node.js client 4.0 supports all data structures available in the CP Subsystem, such as AtomicLong, AtomicReference, FencedLock, Semaphore, and CountDownLatch. Here is how a basic FencedLock usage looks like:

// Get a FencedLock called 'my-lock'
const lock = await client.getCPSubsystem().getLock('my-lock');
// Acquire the lock (returns a fencing token)
const fence = await lock.lock();
try {
    // Your guarded code goes here
} finally {
    // Make sure to release the lock
    await lock.unlock(fence);
}

Backup Acknowledgments

In previous versions, the client was waiting for the sync backups to complete on the member. This was causing 4 network hops to complete a client operation with sync backup. Since sync backup configuration is our out-of-the-box experience, we improved its performance. Backup acknowledgments (a.k.a. boomerang backups) design decreases network hops to 3, thus improving the throughput up to 30%.

Improved Performance

We did a number of experiments and optimizations leading to improved performance for writes by 5-10%.

Other Changes

You can see the list of all changes in this version in the release notes.

What’s Next?

We believe the Node.js client has the capabilities to cover most of your use cases. Next, we are planning to work on integrations with well-known Node.js libraries! Here are the top items in our backlog:

Hazelcast session store for popular Node.js web frameworks: A session store backed by Hazelcast IMDG.
Hazelcast cache adapters for popular ORMs: Hazelcast integration with the Sequelize framework, a promise-based Node.js ORM for SQL databases.
Blue/Green Deployments: Ability to divert the client automatically to another cluster on demand or when the intended cluster becomes unavailable.
Full SQL support: Once the SQL feature in Hazelcast graduated from the beta status, we are going to add it to Node.js client.

You can always check the Hazelcast Node.js client roadmap for an up-to-date list of features in our backlog.

Hazelcast Node.js client 4.0 is available on npm. We look forward to hearing your feedback on our Slack, Stack Overflow, or Google groups. If you would like to introduce some changes or contribute, please visit our Github repository.