Sunday, April 7, 2024

Java 22 Preview: Would Gatherer extension point revolutionize the Stream API in Java ?

Java 22 released few weeks ago and with that comes the gather() method that accepts Gatherer interface in Stream API as a preview feature.

When Stream API was released with Java 8 there was plenty of joy across java developers, but its limited function hid its true potential. Developers often create a list out of the Stream and perform any additional operation imperatively, which often leads to complicated, error-prone code.

To overcome this limitation, with Java 22, Gatherer interface and gather() method introduced to perform complex transformation as 1:1, 1:n, n:1 and n:m.

Now lets go thorough some of the inbuilt Gatherers and its use cases.

Install JDK 22 and enable preview mode by parsing the flag --enable-preview when executing the program.

Gatherers.windowFixed(int)

Window fixed implementation provides a way to group the stream elements in the order into group of fixed number of elements. 

Key points are 

1. No window will be provided if the list is empty

2. Last window may have less number of element than the fixed size given.

Suppose if we have following list of number 1, 2, 3, 4, 5, 6, 7, 8 and if want to group them into group of 3 elements and find the max number among them, then it can be done in this way

var numbers = List.of(1, 2, 3, 4, 5, 6, 7, 8);
var maxNumbers = numbers.stream()
    .gather(Gatherers.windowFixed(3))
    .peek(System.out::println)
    .map(values -> values.stream().max(Comparator.naturalOrder()).orElseThrow())
    .toList();
System.out.println(maxNumbers);

and the result would be

[1, 2, 3]
[4, 5, 6]
[7, 8]
Result [3, 6, 8]

Gatherers.windowSliding(int)

This is similar to window fixed except that it makes the group in sliding one by one, , until the last element is included. 

In case if there are numbers 1, 2, 3 then windowSliding(2) would provide groups as [1,2] and [2, 3]

Key points are 

1. No window provided for empty list

2. If the list contains less elements than sliding size then single window with all the elements provided

3. If the list contains same amount of element or more, than the sliding size then one or more windows where all of its size same as sliding size are provided.

Suppose if we have following list of number 1, 2, 3, 4, 5 and if want to group them into slide of 3 elements and find the max number among them, then it can be done in this way.

var numbers = List.of(1, 2, 3, 4, 5);
var maxNumbers = numbers.stream()
	.gather(Gatherers.windowSliding(3))
	.peek(System.out::println)
	.map(values -> values.stream().max(Comparator.naturalOrder()).orElseThrow())
	.toList();
System.out.println("Result " + maxNumbers);

and the result would be

[1, 2, 3]
[2, 3, 4]
[3, 4, 5]
Result [3, 4, 5]

Gatherers.fold(Supplier, BiFunction)

This is similar to reduce, where a single element is generated from list of element.

Suppose if we have following list of numbers 1, 2, 3, 4, 5 then fold can be used to add all this numbers.

var numbers = List.of(1, 2, 3, 4, 5);
var maxNumbers = numbers.stream()
	.gather(Gatherers.fold(() -> 0, Integer::sum))
	.toList();
System.out.println("Result " + maxNumbers); 

and the result would be

Result [15]

Gatherers.scan(Supplier, BiFunction)

scan is similar to fold except that scan will push the each iteration and fold will push only in the end of the stream.

Suppose if we have following list of numbers 1, 2, 3, 4, 5 and if we want to accumulate the value of each elements with its previous values and get the result as 1, 3, 6, 10, 15.

var numbers = List.of(1, 2, 3, 4, 5);
var maxNumbers = numbers.stream()
	.gather(Gatherers.scan(() -> 0, Integer::sum))
	.toList();
System.out.println("Result " + maxNumbers);

and the result is

Result [1, 3, 6, 10, 15]

More interesting thing about Gatherer is that it brings the possibility of creating our own Gatherer and integrate into Stream.

In the Gatherers.windowFixed example there was a part where max number is found in each window. For that map(values -> values.stream().max(Comparator.naturalOrder()).orElseThrow()) was used and lets try to achieve same thing using our own Gatherer implementation.

public static Gatherer<List<Integer>, ?, Integer> maxNumber() {
	class MaxNumber {
		boolean integrate(List<Integer> elements, Gatherer.Downstream<? super Integer> downstream) {
			var maxValue = elements.stream().max(Comparator.naturalOrder()).orElseThrow();
			return downstream.push(maxValue);
		}
	}

	return Gatherer.ofSequential(
		MaxNumber::new,
		Gatherer.Integrator.<MaxNumber, List<Integer>, Integer>ofGreedy(MaxNumber::integrate)
	);
}
var numbers = List.of(1, 2, 3, 4, 5, 6, 7, 8);
var maxNumbers = numbers.stream()
	.gather(Gatherers.windowFixed(3))
	.gather(maxNumber()) // Our maxNumber gatherer is used
	.toList();
System.out.println("Result " + maxNumbers);

and the result is 

Result [3, 6, 8]

To sum up, Gatherer interface introduced with Stream API looks promising as it allows great level of extendibility to the Stream processing and hopefully would allow us to do many more stream magics in future.

Saturday, February 24, 2024

Use PostgreSQL function to trigger a task on a table upon row insert or update

PostgreSQL provides support to create function which can then be executed on a table upon a specific event.

Suppose if we want to keep only a specific number of latest rows in the table upon each insertion or update, we can use the trigger function to do that, without the need to run any background clean up task.

Lets see how we can do that

Create a table which has id, value and createdDataTime as its columns

create table test (
  id INT PRIMARY KEY, 
  value VARCHAR(50), 
  createdDateTime TIMESTAMP DEFAULT now()
);

then lets create a PostgreSQL procedure function, which upon execution should keep only the last 5 rows created and drop all the others.

CREATE FUNCTION drop_old_records() RETURNS trigger AS 
    '
    BEGIN
        DELETE FROM test WHERE id IN (SELECT id FROM test ORDER BY createdDateTime DESC OFFSET 5);
        RETURN NEW; END; ' LANGUAGE plpgsql;

Now the above created procedure function can be configured to trigger it whenever an insert or update is performed on the table test.

CREATE TRIGGER trigger_on_insert_or_update
AFTER INSERT OR UPDATE
ON test
EXECUTE PROCEDURE drop_old_records();

With that when insert or update performed on the table test then the drop_old_records function will be executed and only the last 5 rows will be kept.

Sample insertions

INSERT INTO test (id, value) VALUES (1, 'one');
INSERT INTO test (id, value) VALUES (2, 'two');
INSERT INTO test (id, value) VALUES (3, 'three');
INSERT INTO test (id, value) VALUES (4, 'four');
INSERT INTO test (id, value) VALUES (5, 'five');
INSERT INTO test (id, value) VALUES (6, 'six');

Result

id	value	createddatetime
2	two	2024-02-24T08:35:23.367Z
3	three	2024-02-24T08:35:23.368Z
4	four	2024-02-24T08:35:23.368Z
5	five	2024-02-24T08:35:23.368Z
6	six	2024-02-24T08:35:23.369Z

Wednesday, January 31, 2024

How to read and process yaml files in POSIX using yq and jq

Prerequisites

yq and jq tools installed

Lets assume we have following yaml file called snapshots.yaml where the fields reference and version should be read for each item in the snapshots key in POSIX script

documentation:
  snapshots:
    - reference: master
      version: latest
    - reference: '2024.1'
      version: '1.2.1'
    - reference: '2023.12'
      version: '1.0.1'

In order to read them in POSIX script lets use the tool yq and jq, it provide support to process yaml and json contents respectively.

Use following command to read documentation.snapshots yaml content as json

yq eval -o=j -I=0 '.documentation.snapshots[]' snapshots.yaml

this will produce the following result, where each line will be a json object

{"reference":"master","version":"latest"}
{"reference":"2024.1","version":"1.2.1"}
{"reference":"2023.12","version":"1.0.1"}

these result can be iterated in POSIX script and each json line can be processed using the jq tool to read the fields reference and version like this.

echo '{"reference":"master","version":"latest"}' | jq -r .reference
echo '{"reference":"master","version":"latest"}' | jq -r .version

With all that the complete POSIX script would look like this.

#!/usr/bin/env sh
set -eu

for snapshot in $(yq eval -o=j -I=0 '.documentation.snapshots[]' snapshots.yaml); do
  reference=$(echo "$snapshot" | jq -r .reference)
  version=$(echo "$snapshot" | jq -r .version)

  echo "Reference: $reference, Version: $version"
done

Result

Reference: master, Version: latest
Reference: 2024.1, Version: 1.2.1
Reference: 2023.12, Version: 1.0.1

Tuesday, January 9, 2024

Create Azure AKS cluster with static egress IP using Terraform

Lets see how we can use a static public IP as a egress IP for the AKS cluster

Prerequisite

Create IP resource

We first need to create a Public IP Address for us to use it in our AKS cluster.

Login to Azure portal  > Create a Resource Group

Choose your subscription and provide a name for the resource group E.g. Name StaticIpExample

Under the newly created StaticIpExample resource group, create a Public IP address resource.

Provide a name. E.g. Name: MyStaticIp, and make sure subscription and resource group are selected to the correct value. Once verified create the resource.

This will create a public IP for you.

Load the Public IP into Terraform

Open aks-cluster.tf file from the Terraform project and add following new block into the file.

data "azurerm_public_ip" "egress" {
  name                = "MyStaticIp"
  resource_group_name = "StaticIpExample"
}

once added, run terraform plan and verify the changes. This change will load the azurerm_public_ip resource named MyStaticIp from StaticIpExample resource group.

Use the static IP in AKS cluster

Now the above loaded static IP should be used in the AKS cluster declaration to use it as egress IP.

Open aks-cluster.tf file and add following block into the resource "azurerm_kubernetes_cluster" "default" {  block.

network_profile {
  network_plugin = "kubenet"
  load_balancer_profile {
    outbound_ip_address_ids = [data.azurerm_public_ip.egress.id]
  }
}

The block sets the kubenet as the default network plugin and sets the already defined static public IP as the outbound IP address for the load balancer.

Provide Access to AKS Cluster

If the Public IP address resource and AKS cluster are in two different resource groups then the AKS cluster needs to be provided with the access to use the IP address in the different resource group.

Create User Assigned Managed Identity resource in the resource group where the Public IP address resource present, in our case StaticIpExample, and provide a name E.g. Name: IpIdentity.

Open Public IP address resource MyStaticIp, navigate to Access Control (IAM) > Add role assignment > choose Contributor role under privileged administrator roles (or choose a role best match for you) > Next > Choose Managed Identity > Select newly created Managed Identity > Next > Review + assign

This will provide access to Managed Identity IpIdentity to handle Public IP address resource MyStaticIp.

Load the created managed identity into Terraform.

Open the aks-cluster.tf file and add following block

data "azurerm_user_assigned_identity" "ip_identity" {
  name                = "IpIdentity"
  resource_group_name = "StaticIpExample"
}

above declared managed identity data should be used in AKS cluster to be able to access the Public IP address resource present in the different resource group.

Open aks-cluster.tf file and add following block into the resource "azurerm_kubernetes_cluster" "default" {  block.

identity {
  type = "UserAssigned"
  identity_ids = [
    data.azurerm_user_assigned_identity.ip_identity.id
  ]
}

that's all and terraform plan then terraform apply should now create an AKS cluster which uses our defined static Ip address as its egress IP. 

This will make sure the cluster egress IP doesn't change when cluster deleted and re-created again.