Primeros pasos con Apache Kafka parte 16

Seguimos con Kafka. 

Existe la necesidad de escalar el consumo por temas. Al igual que varios productores pueden escribir sobre el mismo tema, debemos permitir que varios consumidores lean el mismo tema, dividiendo los datos entre ellos.

Los consumidores de Kafka suelen formar parte de un grupo de consumidores. Cuando varios consumidores están suscritos a un tema y pertenecen al mismo grupo de consumidores, cada consumidor del grupo recibirá mensajes de un subconjunto diferente de las particiones del tema.

Tomemos el tema T1 con cuatro particiones. Ahora suponga que creamos un nuevo consumidor, C1, que es el único consumidor del grupo G1, y lo usamos para suscribirse al tema T1. El consumidor C1 recibirá todos los mensajes de las cuatro particiones t1.

Si agregamos otro consumidor, C2, al grupo G1, cada consumidor solo recibirá mensajes de dos particiones. Quizás los mensajes de la partición 0 y 2 van a C1 y los mensajes de las particiones 1 y 3 van al consumidor C2.

Si G1 tiene cuatro consumidores, cada uno leerá los mensajes de una sola partición.

Si agregamos más consumidores a un solo grupo con un solo tema que las particiones que tenemos, algunos de los consumidores estarán inactivos y no recibirán ningún mensaje.

La principal forma en que escalamos el consumo de datos de un tema de Kafka es agregando más consumidores a un grupo de consumidores. Es común que los consumidores de Kafka realicen operaciones de alta latencia, como escribir en una base de datos o un cálculo lento de los datos. En estos casos, es posible que un solo consumidor no pueda mantenerse al día con los flujos de datos de velocidad en un tema, y agregar más consumidores que compartan la carga al hacer que cada consumidor posea solo un subconjunto de las particiones y los mensajes es nuestro método principal de escalado. Esta es una buena razón para crear temas con una gran cantidad de particiones: permite agregar más consumidores cuando aumenta la carga. Tenga en cuenta que no tiene sentido agregar más consumidores de los que tiene particiones en un tema; algunos de los consumidores simplemente estarán inactivos. 

Además de agregar consumidores para escalar una sola aplicación, es muy común tener múltiples aplicaciones que necesitan leer datos del mismo tema. De hecho, uno de los principales objetivos de diseño en Kafka era hacer que los datos producidos para los temas de Kafka estuvieran disponibles para muchos casos de uso en toda la organización. En esos casos, queremos que cada aplicación obtenga todos los mensajes, en lugar de solo un subconjunto. Para asegurarse de que una aplicación reciba todos los mensajes de un tema, asegúrese de que la aplicación tenga su propio grupo de consumidores. A diferencia de muchos sistemas de mensajería tradicionales, Kafka se adapta a una gran cantidad de consumidores y grupos de consumidores sin reducir el rendimiento. En el ejemplo anterior, si agregamos un nuevo grupo de consumidores G2 con un solo consumidor, este consumidor obtendrá todos los mensajes del tema T1 independientemente de lo que esté haciendo G1. G2 puede tener más de un consumidor, en cuyo caso cada uno obtendrá un subconjunto de particiones, tal como mostramos para G1, pero G2 en su conjunto seguirá recibiendo todos los mensajes independientemente de otros grupos de consumidores.

Para resumir, crea un nuevo grupo de consumidores para cada aplicación que necesita todos los mensajes de uno o más temas. Agrega consumidores a un grupo de consumidores existente para escalar la lectura y el procesamiento de mensajes de los temas, por lo que cada consumidor adicional en un grupo solo obtendrá un subconjunto de los mensajes.

[O'Reilly eBook] Web Application Security

 Me llego un mail con este libro gratuito y quiero compartirlo con ustedes: 

EBOOK While many resources for network and IT security are available, detailed knowledge regarding modern web application security has been lacking – until now. This practical guide provides both offensive and defensive security concepts that software engineers can easily learn and apply. NGINX is proud to make the O’Reilly eBook, Web Application Security, available for free download with our compliments. This eBook is written by Andrew Hoffman, a senior security engineer at Salesforce, and introduces three pillars of web application security: recon, offense, and defense. It also features a foreword by Chris Witeck of NGINX at F5. Download this eBook to learn: About common vulnerabilities plaguing today's web applications How to deploy mitigations to protect your applications against hackers Practical tips to help you improve the overall security of your web applications DOWNLOAD EBOOK

Primeros pasos con Apache Kafka parte 15

Hasta ahora, hemos discutido las características del particionador predeterminado, que es el que se usa con más frecuencia. Sin embargo, Kafka no lo limita a particiones hash y, a veces, existen buenas razones para particionar los datos de manera diferente. Por ejemplo, suponga que es un proveedor B2B y su mayor cliente es una empresa que fabrica dispositivos portátiles llamados Bananas. Suponga que hace tantos negocios con el cliente "Banana" que más del 10% de sus transacciones diarias son con este cliente. Si usa la partición de hash predeterminada, los registros de Banana se asignarán a la misma partición que otras cuentas, lo que dará como resultado que una partición sea aproximadamente el doble de grande que el resto. Esto puede hacer que los servidores se queden sin espacio, que el procesamiento se ralentice, etc. Lo que realmente queremos es darle a Banana su propia partición y luego usar particiones hash para asignar el resto de las cuentas a las particiones.

A continuación, se muestra un ejemplo de un particionador personalizado:

import org.apache.kafka.clients.producer.Partitioner;

import org.apache.kafka.common.Cluster;

import org.apache.kafka.common.PartitionInfo;

import org.apache.kafka.common.record.InvalidRecordException;

import org.apache.kafka.common.utils.Utils;

public class BananaPartitioner implements Partitioner {

public void configure(Map<String, ?> configs) {}

public int partition(String topic, Object key, byte[] keyBytes,

        Object value, byte[] valueBytes,

        Cluster cluster) {

    List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);

    int numPartitions = partitions.size();

    if ((keyBytes == null) || (!(key instanceOf String))) 

        throw new InvalidRecordException("We expect all messages to have customer name as key")

    if (((String) key).equals("Banana"))

        return numPartitions; // Banana siempre va estar en la ultima partición

    return (Math.abs(Utils.murmur2(keyBytes)) % (numPartitions - 1))

}

public void close() {}

}

Creando Observables con RxJava

La forma más sencilla de crear un observable es utilizar los métodos de fabricación que se implementan en la biblioteca RxJava. Ya usamos el método Observable.from(), en un post anterior.  

Observable.just() crea un Observable que emite el objeto o los objetos que se pasan como parámetros:

Observable.just(1, 2, 3, 4, 5)

Observable.range (a, n) crea un Observable que emite un rango de n enteros consecutivos a partir de a :

Observable.range(1, 5)

Observable.interval (long, TimeUnit) : crea un Observable que emite una secuencia de enteros a partir de 0 que están espaciados por un intervalo de tiempo dado. El primer argumento es la cantidad de tiempo y el segundo argumento define la unidad de tiempo. El siguiente observable emite un elemento cada 1 segundo:

Observable.interval(1, TimeUnit.SECONDS)

La secuencia es una secuencia infinital, por lo que onCompleted nunca será notificado. La secuencia se detiene solo cuando no hay más observadores conectados (suscritos) a lo observable. 

Observable.timer (long, TimeUnit) crea un Observable que emite solo un elemento después de un retraso determinado.

Observable.create () es el método permite crear un Observable desde cero. Por ejemplo, si desea crear un observable que emita solo una cadena, "¡Hola!", Puede escribir :

Observable.create(

    new Observable.OnSubscribe<String>() {

        @Override

        public void call(Subscriber<? super String> observer) {

            observer.onNext("Hello!");

            observer.onCompleted();

        }

    }

);

Observable.empty () crea un Observable que emite una secuencia vacía (cero elementos) y luego se completa. Por lo tanto, solo se notificará a onCompleted().

Puede ser útil si desea emitir una secuencia vacía en lugar de emitir elementos nulos o arrojar errores

Observable.error (throwable) crea un Observable que emite una secuencia vacía (cero elementos) y luego notifica un error. Por lo tanto, solo se llamará a onError().

Observable.never () crea un Observable que emite una secuencia vacía (cero elementos) y nunca se completa. No se invocará ningún método del observador.

Observable.defer () crea un Observable solo cuando un suscriptor se suscribe.

La mejor manera de explicar qué hace defer () es con el siguiente ejemplo. Empecemos por la clase Persona, que tiene dos campos: nombre y edad.

class Person {

    private String name;

    private int age;

    // geter y setters

}

Ahora creamos una instancia de Person, y dos Observables para ser notificados con los valores de edad y nombre :

// create a new instance of Person

final Person person = new Person();

Observable<String> nameObservable = Observable.just(person.getName());

Observable<Integer> ageObservable = Observable.just(person.getAge());

// set age and name

person.setName("Bob");

person.setAge(35);

ageObservable.subscribe(new Subscriber<Integer>() {

    @Override

    public void onCompleted() {

    }

    @Override

    public void onError(Throwable e) {

    }

    @Override

    public void onNext(Integer age) {

        System.out.println("age is: " + age);

    }

});

nameObservable.subscribe(new Subscriber<String>() {

    @Override

    public void onCompleted() {

    }

    @Override

    public void onError(Throwable e) {

    }

    @Override

    public void onNext(String name) {

        System.out.println("name is: " + name);

    }

});

¿Qué sucede cuando llama a los métodos observeName () y observeAge () en una instancia de Person? ¿Cuál será la secuencia emitida por los observables? Desafortunadamente, la salida será

age is: 0

name is: null

El problema aquí es que Observable.just() se evalúa tan pronto como se invoca, por lo que creará una secuencia utilizando el valor exacto de ese nombre y referencia de edad cuando se crea el observable. En el ejemplo, cuando se crea el observable, la edad es 0 y el nombre es nulo.

Para esto existe Observable.defer ().

Observable<String> nameObservable = Observable.defer(new

    Func0<Observable<String>>() {

        @Override

        public Observable<String> call() {

            return Observable.just(person.getName());

        }

});

Observable<Integer> ageObservable = Observable.defer(new Func0<Observable<Integer>>() {

    @Override

    public Observable<Integer> call() {

        return Observable.just(person.getAge());

    }

});

Al usar estos dos observables, la salida de los ejemplos anteriores se convierte en

age is: 35

name is: Bob

Observables calientes y fríos

Un Observable comienza a emitir una secuencia de elementos cuando el Observador se suscribe: se denominan observables fríos. Los observables fríos siempre esperan tener al menos un observador suscrito para comenzar a emitir elementos.

Por otro lado, un observable que comienza a emitir elementos antes de conectarse a un observador se denomina observable caliente. Con los observables calientes, un observador puede suscribirse y comenzar a recibir elementos en cualquier momento durante la emisión. Con observables calientes, el observador puede recibir la secuencia completa de elementos comenzando desde el principio o no.

Veamos un ejemplo más concreto, pero simple. 

Creemos un Observable que emita todos los números enteros del 1 al 5 y suscríbase a él:

Observable<Integer> observable = Observable.from(new Integer[]{1, 2, 3, 4, 5});

observable.subscribe(new Subscriber<Integer>() {

    @Override

    public void onCompleted() {

        System.out.println("Sequence completed!");

    }

    @Override

    public void onError(Throwable e) {

        System.err.println("Exception: " + e.getMessage());

    }

    @Override

    public void onNext(Integer integer) {

        System.out.println("next item is: " + integer);

    }

});

La salida esperada es

next item is: 1

next item is: 2

next item is: 3

next item is: 4

next item is: 5

Sequence completed!

Este es un observable frío porque comenzará a emitir elementos solo cuando el observador se suscriba. 

El observable generará una secuencia de cinco elementos, cada uno representando un objeto entero (de 1 a 5), ​​por lo que el método onNext del observador se invocará cinco veces. Al final de la secuencia, se notificará el método onCompleted. El método onError nunca será notificado porque esta secuencia no genera ningún tipo de error o excepción.

Un ejemplo de un observable caliente podría ser un observable que emite un evento cada vez que se hace clic en un botón de la interfaz de usuario. No comienza a emitir eventos cuando el observador se suscribe; emite eventos incluso si no hay ningún suscriptor suscrito. 

En este ejemplo, crea,ps un observable usando el método Observable.from (), un método de fábrica estático que puede crear un Observable a partir de un matriz, iterable o Future.

Esta no es la única forma de crear observables. Pero esa es una historia para otro post...

Primeros pasos con Apache Kafka parte 14

 Seguimos con Kafka. 

Los objetos ProducerRecord incluyen un nombre de tema, una clave y un valor. Los mensajes de Kafka son pares clave-valor y, si bien es posible crear un ProducerRecord con solo un tema y un valor, con la clave establecida en nula de forma predeterminada, la mayoría de las aplicaciones producen registros con claves. Las claves sirven para dos objetivos: son información adicional que se almacena con el mensaje y también se utilizan para decidir en cuál de las particiones de tema se escribirá el mensaje. Todos los mensajes con la misma clave irán a la misma partición. Esto significa que si un proceso está leyendo solo un subconjunto de las particiones en un tema, todos los registros para una sola clave serán leídos por el mismo proceso. Para crear un registro de valor-clave, simplemente cree un ProducerRecord de la siguiente manera:

ProducerRecord<Integer, String> record = new ProducerRecord<>("CustomerCountry", "Laboratory Equipment", "USA");

Al crear mensajes con una clave nula, simplemente puede omitir la clave:

ProducerRecord<Integer, String> record = new ProducerRecord<>("CustomerCountry", "USA");

Aquí, la clave simplemente se establecerá en nula, lo que puede indicar que faltaba el nombre de un cliente en un formulario.

Cuando la clave es nula y se usa el particionador predeterminado, el registro se enviará a una de las particiones disponibles del tema al azar. Se utilizará un algoritmo de operación por turnos para equilibrar los mensajes entre las particiones.

Si existe una clave y se usa el particionador predeterminado, Kafka aplicará un hash a la clave (usando su propio algoritmo hash, por lo que los valores hash no cambiarán cuando se actualice Java) y usará el resultado para asignar el mensaje a una partición específica. Dado que es importante que una clave siempre se asigne a la misma partición, usamos todas las particiones del tema para calcular la asignación, no solo las particiones disponibles. Esto significa que si una partición específica no está disponible cuando escribe datos en ella, es posible que obtenga un error. Esto es bastante raro, como verá en el Capítulo 6 cuando analicemos la replicación y disponibilidad de Kafka.

El mapeo de claves a particiones es consistente solo mientras no cambie el número de particiones en un tema. Entonces, siempre que el número de particiones sea constante, puede estar seguro de que, por ejemplo, los registros relacionados con el usuario 045189 siempre se escribirán en la partición 34. Esto permite todo tipo de optimización al leer datos de particiones. Sin embargo, en el momento en que agrega nuevas particiones al tema, esto ya no está garantizado; los registros antiguos permanecerán en la partición 34, mientras que los registros nuevos se escribirán en una partición diferente. Cuando la partición de claves es importante, la solución más sencilla es crear temas con suficientes particiones y nunca agregar particiones.

onNext, onCompleted, onError

Como vimos anteriormente un flujo de datos lanza diferentes tipos de señales, next si hay un próximo dato y complete si se finalizo el flujo de datos. 

La interfaz rx.Observer <T> define los métodos, onNext, onCompleted, onError(Throwable). onError nos indica que sucedió un error y el flujo de datos no puede trasmitir más. 

Veamos un ejemplo: 

public void subscribeToObservable(Observable<T> observable) {

observable.subscribe(new Subscriber<>() {

@Override

public void onCompleted() {

    // invoked when Observable stops emitting items

}

@Override

public void onError(Throwable e) {

    // invoked when Observable throws an exception

    // while emitting items

}

@Override

public void onNext(T nextItem) {

    // invoked when Observable emits an item

    // usually you will consume the nextItem here

}

});

}

Usamos Subscriber<T> dado que es un objeto que implementa la interfaz rx.Observer <T>. La razón por la que utiliza Subscriber en lugar de cualquier otra implementación de la interfaz de Observer es que el Subscriber también implementa la interfaz de Suscripción, que le permite verificar si el suscriptor está cancelado (con el método isUnsubscriptions ()) y cancelar su suscripción.

En el ejemplo anterior, observe que un observador reacciona a tres tipos de eventos:

• onNext: Ocurre cero, una o más veces. Si la secuencia se completa correctamente, el método onNext se invocará tantas veces como el número de elementos de la secuencia. Si se produce un error en un momento determinado, el método onNext no se invocará más.
• onCompleted: solo cuando todos los elementos de la secuencia se emitan correctamente, se invocará el método onCompleted. Se invoca solo una vez y después de que se haya emitido el último elemento. No va a ser llamado nunca en una secuencia infinita.
• onError: puede ocurrir un error en cada momento de la secuencia, y la secuencia se detendrá inmediatamente. En este caso, se invocará el método onError, pasando el error como objeto Throwable. Los otros dos métodos, onNext y onCompleted, no se invocarán, luego de este. 

Un observable no puede notificar los métodos onCompleted y onError, solo uno de ellos. Siempre será el último método invocado.

Con el método Observable.subscribe () (una operación llamada suscripción), puede conectar un Observable a un Observer, pero ¿qué sucede si desea desconectarlos? Esta la operación llamada unsubscribe:

// disconnect observable and observer

subscription.unsubscribe()

Se puede verificar si la suscripción se ha cancelado con el siguiente método:

subscription.isUnsubscribed()

Luego de la cancelación de suscripción, onNext no recibirá ningún otro elemento y los otros dos métodos, onCompleted y onError, no serán notificados. Después de la cancelación de la suscripción, el observable puede detenerse o continuar con la emisión, pero no se notificará al observador al respecto.

Definiendo Observable y Observer

En Reactive programming un Observable es un objeto que emite una secuencia (o flujo) de eventos. Representa una colección basada en inserción, que es una colección en la que se insertan eventos cuando se crean.

Un observable emite una secuencia que puede ser vacía, finita o infinita. Cuando la secuencia es finita, se emite un evento completo después del final de la secuencia. En cualquier momento durante la emisión (pero no después de su finalización) se puede emitir un evento de error, deteniendo la emisión y cancelando la emisión del evento completo.

Cuando la secuencia está vacía, solo se emite el evento completo, sin emitir ningún ítem. Con una secuencia infinita, el evento completo nunca se emite.

La emisión se puede transformar, filtrar o combinar con otras emisiones, etc. 

Un observador es un objeto que se suscribe a un observable. Escucha y reacciona a cualquier secuencia de elementos emitida por el Observable.

El Observer no está bloqueado mientras espera nuevos elementos emitidos, por lo que en operaciones simultáneas, no se produce ningún bloqueo. Simplemente se activa cuando se emite un nuevo elemento.

Este es uno de los principios fundamentales de la programación reactiva: en lugar de ejecutar instrucciones una a la vez (siempre esperando a que se complete la instrucción anterior), el observable proporciona un mecanismo para recuperar y transformar datos, y el Observer activa este mecanismo, todos de forma concurrente.

El siguiente pseudocódigo es un ejemplo del método que implementa el Observer que reacciona a los elementos del Observable:

onNext = { it -> doSomething }

Aquí, el método está definido, pero no se invoca nada. Para comenzar a reaccionar, debe suscribirse al Observable:

observable.subscribe(onNext)

Ahora el observador está atento a los elementos y reaccionará a cada elemento nuevo que se emitirá.

Reescribamos este ejemplo en código Java usando las API de RxJava:

public void subscribeToObservable(Observable<T> observable) {

    observable.subscribe(nextItem -> {

        // invoked when Observable emits an item

        // usually you will consume the nextItem here

    });

}

Ahora está claro que para conectar un observable con un observador, debes usar el método de suscripción.

Creando la primera aplicación con Quarkus parte 5

Seguimos con quarkus

Vamos a guardar unos saludos para luego poder recuperarlos y usarlos para saludar. 

Para esto vamos a crear una clase con el saludo : 

package com.hexacta.model;

import javax.persistence.*;
import java.util.Objects;

@Entity
public class Greeting {

    @GeneratedValue(strategy = GenerationType.IDENTITY)
    @Column(nullable = false)
    @Id
    private Integer id;

    @Column
    private String value;

    public void setId(Integer id) {
        this.id = id;
    }

    public Integer getId() {
        return id;
    }

    public Greeting() {}

    public Greeting(String value) {
        this.value = value;
    }

    public String getValue() {
        return value;
    }

    public void setValue(String value) {
        this.value = value;
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
        Greeting greeting = (Greeting) o;
        return Objects.equals(value, greeting.value);
    }

    @Override
    public int hashCode() {
        return Objects.hash(value);
    }
}

Luego vamos a hacer una clase DAO para acceso a la base de datos : 

package com.hexacta.dao;

import com.hexacta.model.Greeting;

import javax.enterprise.context.ApplicationScoped;
import javax.inject.Inject;
import javax.persistence.EntityManager;

@ApplicationScoped
public class GreetingDAO {

    @Inject
    private EntityManager em;

    public int save(Greeting aGreeting) {
        this.em.persist(aGreeting);
        return aGreeting.getId();
    }

    public Greeting get(int id) {
        var qr = em.createQuery("from com.hexacta.model.Greeting g " +
                "where g.id = ?1");
        qr.setParameter(1, id);
        return (Greeting) qr.getSingleResult();
    }
}

Modificamos el servicio para que permita guardar los saludos : 

package com.hexacta;

import com.hexacta.dao.GreetingDAO;
import com.hexacta.model.Greeting;

import javax.transaction.Transactional;
import javax.enterprise.context.ApplicationScoped;
import javax.inject.Inject;

@ApplicationScoped
public class GreetingServices {

    @Inject
    private GreetingDAO dao;

    public String greeting(String name) {
        return "hola " + name;
    }

    public String greeting(int id,String name) {
        Greeting aGreeting = dao.get(id);
        if (aGreeting == null) {
            return name;
        }
        return aGreeting.getValue() + " " + name;
    }

    @Transactional
    public int saveGreeting(String greeting) {
        Greeting aGreeting = new Greeting(greeting);
        return dao.save(aGreeting);
    }
}

Y por último vamos a modificar los servicios REST : 

package com.hexacta;

import org.jboss.resteasy.annotations.jaxrs.PathParam;

import javax.annotation.PostConstruct;
import javax.inject.Inject;
import javax.ws.rs.GET;
import javax.ws.rs.POST;
import javax.ws.rs.Path;
import javax.ws.rs.Produces;
import javax.ws.rs.core.MediaType;

@Path("/hello")
public class GreetingResource {

    private GreetingServices service;

    @Inject
    public GreetingResource(GreetingServices service) {
        this.service = service;
    }

    @GET
    @Produces(MediaType.TEXT_PLAIN)
    public String hello() {
        return "Hello RESTEasy";
    }

    @GET
    @Produces(MediaType.TEXT_PLAIN)
    @Path("/{name}")
    public String hello(@PathParam String name) {
        return this.service.greeting(name);
    }

    @POST
    @Path("/save/{greeting}")
    public int saveGreeting(@PathParam String greeting) {
        return this.service.saveGreeting(greeting);
    }

    @GET
    @Produces(MediaType.TEXT_PLAIN)
    @Path("/{id}/{name}")
    public String hello(@PathParam int id,@PathParam String name) {
        return this.service.greeting(id, name);
    }


}

Y tengo que configurar h2 que es la base que estamos usando, en el application.properties : 

# datasource configuration
quarkus.datasource.db-kind = h2
quarkus.datasource.username = sa
quarkus.datasource.password = a
quarkus.datasource.jdbc.url = jdbc:h2:~/quarkus.db

quarkus.hibernate-orm.database.generation=update

Y listo, podemos guardar diferentes saludos y usarlos. 

Dejo el link del repo : 

https://github.com/emanuelpeg/quarkusExample/

Creando la primera aplicación con Quarkus parte 4

Seguimos con quarkus

Ahora vamos a empaquetar nuestra app, lo hacemos con gradle jar y la ejecutamos con 

java -jar build/quarkus-app/quarkus-run.jar

y listo tenemos nuestra app andando en http://localhost:8080/hello 

El directorio quarkus-app que contiene el archivo jar quarkus-run.jar, que es un jar ejecutable. Pero no tiene todas las dependencias estas se copian en subdirectorios de quarkus-app/lib /. Por lo tanto debemos deployar todo el directorio quarkus-app la primera vez y cada vez que hay cambio de librerías. 

Bueno, ya estamos! con esto tenemos nuestra app andando pero vamos a hacer un paso más vamos a hacer nuestra applicación nativa. Para eso debemos utilizar GraalVM 11. 

La construcción de un ejecutable nativo requiere el uso de una distribución de GraalVM. Hay tres distribuciones: Oracle GraalVM Community Edition (CE), Oracle GraalVM Enterprise Edition (EE) y Mandrel. Las diferencias entre las distribuciones de Oracle y Mandrel son las siguientes:

Mandrel es una distribución descendente de Oracle GraalVM CE. El objetivo principal de Mandrel es proporcionar una forma de crear ejecutables nativos diseñados específicamente para admitir Quarkus.

Las versiones de Mandrel se crean a partir de una base de código derivada de la base de código anterior de Oracle GraalVM CE, con solo cambios menores pero algunas exclusiones importantes que no son necesarias para las aplicaciones nativas de Quarkus. Admiten las mismas capacidades para crear ejecutables nativos que Oracle GraalVM CE, sin cambios significativos en la funcionalidad. En particular, no incluyen soporte para programación políglota. El motivo de estas exclusiones es proporcionar un mejor nivel de soporte para la mayoría de los usuarios de Quarkus. Estas exclusiones también significan que Mandrel ofrece una reducción considerable en su tamaño de distribución en comparación con Oracle GraalVM CE/EE.

Mandrel está construido de forma ligeramente diferente a Oracle GraalVM CE, utilizando el proyecto estándar OpenJDK. Esto significa que no se beneficia de algunas pequeñas mejoras que Oracle ha agregado a la versión de OpenJDK utilizada para crear sus propias descargas de GraalVM. Estas mejoras se omiten porque OpenJDK no las gestiona y no puede responder por ellas. Esto es particularmente importante cuando se trata de conformidad y seguridad.

Actualmente, Mandrel solo se recomienda para compilar ejecutables nativos destinados a entornos Linux en contenedores. Esto significa que los usuarios de Mandrel deben usar contenedores para construir sus ejecutables nativos. Si está creando ejecutables nativos para plataformas de destino macOS o Windows, debería considerar usar Oracle GraalVM en su lugar, porque Mandrel no se dirige actualmente a estas plataformas. Es posible compilar ejecutables nativos directamente en Linux.

Los requisitos previos varían ligeramente dependiendo de si está utilizando Oracle GraalVM CE/EE o Mandrel.

Primero, tenemos que configurar las variables de entorno JAVA_HOME y GRAALVM_HOME (y tener instalado docker). 

Y luego tenemos que hacer : 

gradel build -Dquarkus.package.type=native

docker build -f src/main/docker/Dockerfile.native -t quarkus/demo .

docker run -i --rm -p 8080:8080 quarkus/demo

Cuando lo corrí la primera vez, me tiro error porque estoy usando Windows y me dijo que instale el generador de imágenes windows haciendo esto : 

gu install native-image

 

Primeros pasos con Apache Kafka parte 13

Seguimos con Kafka. 

Los archivos Avro deben almacenar el esquema completo en el archivo de datos que están asociados, almacenar el esquema completo en cada registro generalmente será más del doble del tamaño del registro. Para resolver este problema podemos utilizar un registro de esquemas. Los Schema Registry no forma parte de Apache Kafka, pero hay varias opciones de código abierto para elegir. Usaremos Confluent Schema Registry para este ejemplo. 

La idea es almacenar todos los esquemas utilizados luego, simplemente almacenamos el identificador del esquema en el registro que producimos en Kafka. Los consumidores pueden usar el identificador para extraer el registro del registro de esquema y deserializar los datos. 

A continuación, se muestra un ejemplo de cómo producir objetos Avro generados en Kafka:

Properties props = new Properties();

props.put("bootstrap.servers", "localhost:9092");

props.put("key.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");

props.put("value.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");

props.put("schema.registry.url", schemaUrl);

String topic = "customerContacts";

int wait = 500;

Producer<String, Customer> producer = new KafkaProducer<String,Customer>(props);

while (true) {

    Customer customer = CustomerGenerator.getNext();

    System.out.println("Generated customer " + customer.toString());

    ProducerRecord<String, Customer> record = new ProducerRecord<>(topic, customer.getId(), customer);

    producer.send(record);

}

Usamos KafkaAvroSerializer para serializar nuestros objetos con Avro. AvroSerializer también puede manejar primitivas, por lo que luego podemos usar String como clave de registro y nuestro objeto Customer como valor.

schema.registry.url es un nuevo parámetro. Esto simplemente apunta a dónde almacenamos los esquemas.

El cliente es nuestro objeto generado. Le decimos al productor que nuestros registros contendrán Cliente como valor.

También creamos una instancia de ProducerRecord con Customer como el tipo de valor y pasamos un objeto Customer al crear el nuevo registro.

Eso es todo. Enviamos el registro con nuestro objeto Cliente y KafkaAvroSerializer se encargará del resto.

¿Qué sucede si prefiere utilizar objetos Avro genéricos en lugar de los objetos Avro generados?

No hay problema. En este caso, solo necesita proporcionar el esquema:

Properties props = new Properties();

props.put("bootstrap.servers", "localhost:9092");

props.put("key.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");

props.put("value.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");

props.put("schema.registry.url", url);

String schemaString = "{\"namespace\": \"customerManagement.avro\",

\"type\": \"record\", " +

"\"name\": \"Customer\"," +

"\"fields\": [" +

"{\"name\": \"id\", \"type\": \"int\"}," +

"{\"name\": \"name\", \"type\": \"string\"}," +

"{\"name\": \"email\", \"type\": [\"null\",\"string

\"], \"default\":\"null\" }" +

"]}";

Producer<String, GenericRecord> producer = new KafkaProducer<String, GenericRecord>(props);

Schema.Parser parser = new Schema.Parser();

Schema schema = parser.parse(schemaString);

for (int nCustomers = 0; nCustomers < customers; nCustomers++) {

    String name = "exampleCustomer" + nCustomers;

    String email = "example " + nCustomers + "@example.com"

    GenericRecord customer = new GenericData.Record(schema);

    customer.put("id", nCustomer);

    customer.put("name", name);

    customer.put("email", email);

    ProducerRecord<String, GenericRecord> data = new ProducerRecord<String,

    GenericRecord>("customerContacts", name, customer);

    producer.send(data);

}

Seguimos usando el mismo KafkaAvroSerializer.

Y proporcionamos el URI del mismo registro de esquema.

Pero ahora también necesitamos proporcionar el esquema Avro, ya que no lo proporciona el objeto generado por Avro.

Nuestro tipo de objeto es un Avro GenericRecord, que inicializamos con nuestro esquema y los datos que queremos escribir.

Entonces, el valor de ProducerRecord es simplemente un GenericRecord que cuenta nuestro esquema y datos. El serializador sabrá cómo obtener el esquema de este registro, almacenarlo en el registro de esquema y serializar los datos del objeto.

Primeros pasos con Apache Kafka parte 12

Seguimos con Kafka. 

Apache Avro es un formato de serialización de datos independiente del lenguaje. El proyecto fue creado por Doug Cutting para proporcionar una forma de compartir archivos de datos con una gran audiencia.

Los datos de Avro se describen en un esquema independiente del lenguaje. El esquema generalmente se describe en JSON y la serialización suele ser en archivos binarios, aunque también se admite la serialización en JSON. Avro asume que el esquema está presente al leer y escribir archivos, generalmente incrustando el esquema en los propios archivos.

Una de las características más interesantes de Avro, y lo que lo hace adecuado para su uso en sistemas de mensajería como Kafka, es que cuando la aplicación que está escribiendo mensajes cambia a un nuevo esquema, las aplicaciones que leen los datos pueden continuar procesando mensajes sin necesidad de cambiar o actualizar.

Supongamos que el esquema original fuera:

{

     "namespace": "customerManagement.avro",

     "type": "record",

     "name": "Customer",

     "fields": [

          {"name": "id", "type": "int"},

          {"name": "name", "type": "string""},

          {"name": "faxNumber", "type": ["null", "string"], "default": "null"}

     ]

}

Usamos este esquema durante unos meses y generamos algunos terabytes de datos en este formato. Ahora suponga que decidimos que en la nueva versión, actualizaremos al siglo XXI y ya no incluiremos un campo de número de fax y en su lugar usaremos un campo de correo electrónico.

El nuevo esquema sería:

{"namespace": "customerManagement.avro",

    "type": "record",

    "name": "Customer",

    "fields": [

        {"name": "id", "type": "int"},

        {"name": "name", "type": "string"},

        {"name": "email", "type": ["null", "string"], "default": "null"}

    ]

}

Ahora, después de actualizar a la nueva versión, los registros antiguos contendrán "faxNumber" y los registros nuevos contendrán "email". En muchas organizaciones, las actualizaciones se realizan lentamente y durante muchos meses. Por lo tanto, debemos considerar cómo las aplicaciones anteriores a la actualización que aún usan los números de fax y las aplicaciones posteriores a la actualización que usan el correo electrónico podrán manejar todos los eventos en Kafka.

La aplicación de lectura contendrá llamadas a métodos similares a getName (), getId () y getFaxNumber. Si encuentra un mensaje escrito con el nuevo esquema, getName() y getId () continuará funcionando sin modificaciones, pero getFaxNumber () devolverá nulo porque el mensaje no contendrá un número de fax.

Ahora suponga que actualizamos nuestra aplicación de lectura y ya no tiene el método getFaxNumber() sino getEmail(). Si encuentra un mensaje escrito con el esquema anterior, getEmail() devolverá un valor nulo porque los mensajes anteriores no contienen una dirección de correo electrónico.

Este ejemplo ilustra el beneficio de usar Avro: aunque cambiemos el esquema en los mensajes sin cambiar todas las aplicaciones que leen los datos, no habrá excepciones ni errores de ruptura y no será necesario realizar costosas actualizaciones de los datos existentes.

Sin embargo, hay dos advertencias para este escenario:

• El esquema utilizado para escribir los datos y el esquema esperado por la aplicación de lectura deben ser compatibles. La documentación de Avro incluye reglas de compatibilidad.
• El deserializador necesitará acceder al esquema que se utilizó al escribir los datos, incluso cuando sea diferente del esquema esperado por la aplicación que accede a los datos. En los archivos Avro, el esquema de escritura se incluye en el propio archivo, pero hay una mejor manera de manejar esto para los mensajes de Kafka. Que veremos en próximos post...

Fue lanzado Java 17!!

Oracle ha lanzado la versión 17 del lenguaje de programación Java. Como el primer lanzamiento como long-term support (LTS) desde JDK 11 en 2018.

Puff como pasan las versiones de java, imparable. Como características nuevas tenemos : 

  306: Restore Always-Strict Floating-Point Semantics

  356: Enhanced Pseudo-Random Number Generators

  382: New macOS Rendering Pipeline

  391: macOS/AArch64 Port

  398: Deprecate the Applet API for Removal

  403: Strongly Encapsulate JDK Internals

  406: Pattern Matching for switch (Preview)

  407: Remove RMI Activation

  409: Sealed Classes

  410: Remove the Experimental AOT and JIT Compiler

  411: Deprecate the Security Manager for Removal

  412: Foreign Function & Memory API (Incubator)

  414: Vector API (Second Incubator)

  415: Context-Specific Deserialization Filters

Bueno, este post es para hacerme eco de la noticia, vamos a ir probando más adelante...

Dejo link : https://jdk.java.net/17/

ReactiveX

ReactiveX es un framework para los lenguajes de programación más utilizados: Java, JavaScript, C#, Scala, Clojure, C ++, Ruby, Python, Groovy, JRuby, Kotlin, Swift y más. 

RxJava es un framwork que implementa los conceptos de ReactiveX en Java. Veamos un ejemplo RxJava:

List<Integer> input = Arrays.asList(1, 2, 3, 4, 5);

Observable.from(input).filter(new Func1() {

    @Override

     public Boolean call(Integer x) {

          return x % 2 == 0;

     }

})

o usando lambda : 

Observable.from(input).filter(x -> x % 2 == 0);

El objeto resultante (la instancia de rx.Observable) generará una secuencia de los números pares contenidos en la secuencia de entrada: 2 y 4.

En RxJava, rx.Observable agrega dos semánticas al patrón Observador de Gang of Four (la semántica predeterminada es emitir elementos creados, como una lista con elementos 2,4 en el ejemplo anterior):

• El productor puede notificar al consumidor que no hay más datos disponible.
• El productor puede notificar al consumidor que ha ocurrido un error.

La biblioteca RxJava proporciona un modelo de programación donde podemos trabajar con eventos generados desde UI o llamadas asincrónicas de la misma manera en que operamos con colecciones y streams en Java 8.

La biblioteca RxJava se creó en Netflix como una alternativa más inteligente a Java Futures y devoluciones de llamada. Tanto los futuros como las devoluciones de llamada son fáciles de usar cuando solo hay un nivel de ejecución asincrónica, pero son difíciles de administrar cuando están anidados.

El siguiente ejemplo muestra cómo se maneja el problema de las devoluciones de llamada anidadas en RxJava.

Suponga que necesita llamar a una API remota para autenticar a un usuario, luego a otra para obtener los datos del usuario y a otra API para obtener los contactos de un usuario. Normalmente, tendría que escribir llamadas a API anidadas y hacer complejos callbacks. Pero con RxJava se puede hacer así : 

serviceEndpoint.login().doOnNext(accessToken -> storeCredentials(accessToken)).flatMap(accessToken -> serviceEndpoint.getUser()).flatMap(user -> serviceEndpoint.getUserContact(user.getId()))

Primeros pasos con Apache Kafka parte 11

Seguimos con Kafka. 

Como se vio en ejemplos anteriores, la configuración del productor incluye serializadores y hemos visto cómo utilizar el serializador de cadenas predeterminado. Kafka también incluye serializadores para enteros y ByteArrays, pero algunas veces necesitamos serializar de una forma especial.

Cuando el objeto que necesita enviar a Kafka no es una simple cadena o un entero, tiene la opción de usar una biblioteca de serialización genérica como Avro, Thrift o Protobuf para crear registros, o crear una serialización personalizada para los objetos que ya está usando . 

Suponga que en lugar de registrar solo el nombre del cliente, crea una clase simple para representar a los clientes:

public class Customer {

      private int customerID;

      private String customerName;

      public Customer(int ID, String name) {

            this.customerID = ID;

            this.customerName = name;

      }

      public int getID() {

            return customerID;

      }

      public String getName() {

            return customerName;

      }

}

Ahora suponga que queremos crear un serializador personalizado para esta clase.:

import org.apache.kafka.common.errors.SerializationException;

import java.nio.ByteBuffer;

import java.util.Map;

public class CustomerSerializer implements Serializer<Customer> {

      @Override

      public void configure(Map configs, boolean isKey) {

            // nothing to configure

      }

      @Override

      /**

      We are serializing Customer as:

      4 byte int representing customerId

      4 byte int representing length of customerName in UTF-8 bytes (0 if name is Null)

      N bytes representing customerName in UTF-8

      */

      public byte[] serialize(String topic, Customer data) {

      try {

            byte[] serializedName;

            int stringSize;

            if (data == null)

                  return null;

            else {

                  if (data.getName() != null) {

                        serializeName = data.getName().getBytes("UTF-8");

                        stringSize = serializedName.length;

                  } else {

                        serializedName = new byte[0];

                        stringSize = 0;

                  }

            }

            ByteBuffer buffer = ByteBuffer.allocate(4 + 4 + stringSize);

            buffer.putInt(data.getID());

            buffer.putInt(stringSize);

            buffer.put(serializedName);

            return buffer.array();

      } catch (Exception e) {

            throw new SerializationException("Error when serializing Customer to byte[] " + e);

      }

   }

  @Override

  public void close() {

         // nothing to close

  }

}

La configuración de un productor con este CustomerSerializer le permitirá definir ProducerRecord <String, Customer> y enviar datos del cliente y pasar los objetos del cliente directamente al productor. 

Este ejemplo es bastante simple, pero puede ver lo frágil que es el código. Si alguna vez tenemos demasiados clientes, por ejemplo, y necesitamos cambiar customerID a Long, o si alguna vez decidimos agregar un campo startDate a Customer, tendremos un problema serio para mantener la compatibilidad entre los mensajes antiguos y nuevos. La depuración de problemas de compatibilidad entre diferentes versiones de serializadores y deserializadores es bastante desafiante; es necesario comparar matrices de bytes sin procesar. Para empeorar las cosas, si varios equipos de la misma empresa terminan escribiendo datos del Cliente en Kafka, todos deberán usar los mismos serializadores y modificar el código al mismo tiempo.

Por estos motivos, es buena idea utilizar serializadores y deserializadores existentes, como JSON, Apache Avro, Thrift o Protobuf.