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. 

Libro Gratuito sobre microservicios

EBOOK Microservices architectures offer faster change speeds, better scalability, and cleaner, evolvable system designs. But implementing your first microservices architecture is difficult. How do you make myriad choices, educate your team on all the technical details, and lead the organization to a successful execution to maximize your chance of success? With this book, complements of NGINX, authors Ronnie Mitra and Irakli Nadareishvili provide step-by-step guidance for building an effective microservices architecture. They guide you on an implementation journey based on techniques and architectures that have proven to work for microservices systems – building an operating model, a microservices design, an infrastructure foundation, and two working microservices, then putting those pieces together as a single implementation. For anyone tasked with building microservices or a microservices architecture, this guide is invaluable. In this eBook you will learn: How to design an effective and explicit microservices system end-to-end About forming teams, assigning responsibilities, and working together How to slice a big application into a collection of microservices About building a simple yet powerful CI/CD pipeline for infrastructure changes Download eBook

El patrón observador en la programación reactiva

El patrón Observer es un patrón de diseño en el que hay dos tipos de objetos: observadores y sujetos observables. Un observador es un objeto que observa los cambios de uno o más sujetos; un sujeto es un objeto que mantiene una lista de sus observadores y les notifica automáticamente cuando cambia de estado.

La definición del patrón Observer del libro "Gang of Four" (Design Patterns: Elements of Reusable Object-Oriented Software por Erich Gamma, Richard Helm, Ralph Johnson y John Vlissides, ISBN 0-201-63361-2) es "Defina una dependencia de uno a varios entre objetos para que cuando un objeto cambie de estado, todos sus dependientes sean notificados y actualizados automáticamente".

Este patrón es el núcleo de la programación reactiva. Se ajusta perfectamente al concepto de programación reactiva al proporcionar las estructuras para implementar el mecanismo produce/react

Java SDK implementa el patrón Observer con la clase java.util.Observable y la interfaz java.util.Observer.

class Subject extends java.util.Observable {

    public void doWorkAndNotify() {

        Object result = doWork();

        notifyObservers(result);

    }

}

class MyObserver implements Observer {

    @Override

    public void update(Observable obs, Object item) {

        doSomethingWith(item)

    }

}

La clase Subject extiende java.utils.Observable y es responsable de producir un objeto y notificar a los observadores tan pronto como se haya producido el elemento.

MyObserver implementa Observer y es responsable de observar al sujeto y consumir todos los elementos que produce el sujeto.

Veamos como interactuan estos objetos: 

MyObserver myObserver = new MyObserver();

Subject subject = new Subject();

subject.addObserver(myObserver);

subject.doWorkAndNotify();

Desafortunadamente, esta implementación se revela demasiado simple cuando comienza a escribir una lógica más compleja. No se utilizan estas implementaciones.

Programación reactiva

La programación reactiva lleva la programación funcional un poco más allá, agregando el concepto de flujo de datos y propagación de cambios de datos.

En programación imperativa, se puede asignar un valor a una variable de la siguiente manera:

x = y + z

Aquí, la suma de y+z se asignará a la variable x , si llamamos esta función, los cambios que ocurran a y y a z no modificaran a x.

En la programación reactiva, el valor de x debe actualizarse siempre que cambien los valores de y o z.

Entonces, si los valores iniciales son y = 1 y z = 1, tendrás

x = y + z = 2.

Si y (o z) cambia su valor, esto no significa que x cambia automáticamente, pero debe implementar un mecanismo para actualizar los valores de x cuando se cambian los valores de y y z.

La programación reactiva funcional es un nuevo paradigma de programación; Erik Meijer lo hizo popular (quien creó la biblioteca Rx para .NET cuando trabajaba en Microsoft) y se basa en dos conceptos:

• El código "reacciona" a los eventos.
• El código maneja los valores a medida que varían en el tiempo, propagando los cambios a cada parte del código que usa esos valores.

La clave para entender la programación reactiva es pensar en ella como si operara en un flujo de datos.

Pero, ¿qué quiero decir con "flujo de datos"? Se refiere a una secuencia de eventos, donde un evento podría ser una entrada del usuario (como un toque en un botón), una respuesta de una solicitud de API (como un feed de Facebook), datos contenidos en una colección o incluso una sola variable.

En la programación reactiva, a menudo hay un componente que actúa como fuente, emitiendo una secuencia de elementos (o un flujo de datos) y algunos otros componentes que observan este flujo de elementos y reaccionan a cada elemento emitido ("reaccionan" al elemento emisión).

Quarkus para desarrolladores Spring

Quiero recomendarles este libro gratuito Quarkus for Spring Developers de la gente de Red hat. 

Quarkus for Spring Developers presenta Quarkus a los desarrolladores de Java, con un ojo especial en ayudar a quienes están familiarizados con las convenciones de Spring a realizar una transición rápida y sencilla.

Y sin más dejo link: 

https://developers.redhat.com/books/quarkus-for-spring-developers?sc_cid=7013a000002ph3pAAA

Creando la primera aplicación con Quarkus parte 3

Seguimos con quarkus

quarkus:dev ejecuta Quarkus en modo desarrollo. Esto permite deployment en caliente con compilación en segundo plano, lo que significa que cuando modifica sus archivos Java y / o sus archivos de recursos y actualiza su navegador, estos cambios se aplicarán automáticamente. Esto también funciona para archivos de recursos como el archivo de propiedades de configuración. La actualización del navegador desencadena un análisis del espacio de trabajo y, si se detecta algún cambio, los archivos Java se vuelven a compilar y la aplicación se vuelve a implementar; su solicitud luego es atendida por la aplicación redistribuida. Si hay algún problema con la compilación o la implementación, una página de error se lo informará.

Esto también escuchará un depurador en el puerto 5005. Si desea esperar a que el depurador se conecte antes de ejecutarse, puede pasar -Dsuspend en la línea de comandos. Si no desea el depurador en absoluto, puede usar -Ddebug = false.

En mi caso quiero hacer debug con intellij pero no funciona, si el deploy en caliente pero no me funciona el debug llamando a gradle quarkus:dev en debug.  Lo que hay que hacer es muy fácil, ir a run -> attach to process y en ese menú se va a listar el puerto 5005 que es el puerto de debug y luego el debug anda joya. 

Ya tenemos el debug y el deployment en caliente. Ahora vamos a ver los test, los test unitarios tienen que estar y escapa a este post y a la tecnología que usemos. Ahora bien, esta bueno hacer test de integración para poder revisar toda mi aplicación. desde el servicio REST a la base. Para esto, quarkus nos crea un test por defecto : 

@QuarkusTest

public class GreetingResourceTest {

    @Test

    public void testHelloEndpoint() {

        given()

          .when().get("/hello")

          .then()

             .statusCode(200)

             .body(is("Hello RESTEasy"));

    }

}

Con esto estamos probando que el servicio /hello retorne 200 y su body sea "Hello RESTEasy"

Si corremos el test, quarqus nos levanta el server y le pega. 

Vamos a hacer un test de nuestro servicio /hello/mundo 

    @Test

    public void testHelloWithParameterEndpoint() {

        var name = "Mundo";

        given()

                .when().get("/hello/" + name)

                .then()

                .statusCode(200)

                .body(is("hola " + name));

    }

Y listo!! 

 

Creando la primera aplicación con Quarkus parte 2

 Seguimos con quarkus

Ahora vamos a tomar la app del post pasado y vamos a hacer un servicio que salude y lo vamos a inyectar y luego creamos un web services REST que permita utilizarlo. 

import javax.enterprise.context.ApplicationScoped;

@ApplicationScoped

public class GreetingServices {

    public String greeting(String name) {

        return "holas " + name;

    }

}

Y ahora inyectamos : 

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

import javax.inject.Inject;

import javax.ws.rs.GET;

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);

    }

}

Siempre que podamos debemos inyectar a nivel de constructor, de esta manera queda más explicita la dependencia. 

Y listo!! si vamos a http://localhost:8080/hello/mundo vamos a obtener : 

holas mundo

Vamos a seguir en este post : https://emanuelpeg.blogspot.com/2021/09/creando-la-primera-aplicacion-con_6.html

Creando la primera aplicación con Quarkus

Empecemos por el principio, necesitamos graalvm 11 o superior (en realidad cualquier Jdk 11 o superior pero yo voy a utilizar graalvm para sacar todo el jugo) , maven o gradle (yo voy a usar intellij que viene con gradle y listo) y un ide. 

Pueden usar maven o gradle para crear el proyecto o intellij (tambien) pero yo utilice la pagina https://code.quarkus.io/ y luego lo abrí con intellij. Pero eso lo hice porque quise, pueden hacerlo como quieran. 

Más allá de si eligieron maven o gradle o como hiciero para crear el proyecto, deberían tener esta dependencia : 

implementation 'io.quarkus:quarkus-resteasy'

Ahora veamos el código que genero, es un hola mundo común en un servicio REST : 

import javax.ws.rs.GET;

import javax.ws.rs.Path;

import javax.ws.rs.Produces;

import javax.ws.rs.core.MediaType;

@Path("/hello")

public class GreetingResource {

    @GET

    @Produces(MediaType.TEXT_PLAIN)

    public String hello() {

        return "Hello RESTEasy";

    }

}

Si ejecutamos esto con : 

gradle :quarkusDev 

Va a demorar un rato pero luego si vamos a http://localhost:8080/hello vemos : 

Hello RESTEasy

Y como primer acercamiento, bien, en proximos post vamos a seguir desarrollando este ejemplo. 

Primeros pasos con Apache Kafka parte 10

 

Seguimos con Kafka. 

Hasta ahora, hemos visto muy pocos parámetros de configuración para los productores, solo el URI y los serializadores de bootstrap.servers obligatorios. El productor tiene una gran cantidad de parámetros de configuración; la mayoría están documentados en la documentación de Apache Kafka y muchos tienen valores predeterminados razonables, por lo que no hay razón para modificar cada parámetro. Sin embargo, algunos de los parámetros tienen un impacto significativo en el uso de la memoria, el rendimiento y la confiabilidad de los productores. Los revisaremos aquí.

acks

El parámetro acks controla cuántas réplicas de partición deben recibir el registro antes de que el productor pueda considerar que la escritura se ha realizado correctamente. Esta opción tiene un impacto significativo en la probabilidad de que se pierdan los mensajes. Hay tres valores permitidos para el parámetro acks:

• Si acks = 0, el productor no esperará una respuesta del broker antes de asumir que el mensaje se envió correctamente. Esto significa que si algo salió mal y el corredor no recibirá mensaje, el productor no lo sabrá y el mensaje se perderá. Sin embargo, debido a que el productor no está esperando respuesta del servidor, puede enviar mensajes tan rápido como lo admita la red, por lo que esta configuración se puede utilizar para lograr un rendimiento muy alto.

• Si acks = 1, el productor recibirá una respuesta de éxito del corredor en el momento en que la réplica del líder reciba el mensaje. Si el mensaje no se puede escribir al líder (por ejemplo, si el líder colapsó y aún no se eligió un nuevo líder), el productor recibirá una respuesta de error y podrá volver a intentar enviar el mensaje, evitando la pérdida potencial de datos. El mensaje aún puede perderse si el líder falla y se elige una réplica sin este mensaje como el nuevo líder (a través de unclean elección de líder). En este caso, el rendimiento depende de si enviamos mensajes de forma sincrónica o asincrónica. Si nuestro código de cliente espera una respuesta del servidor (al llamar al método get () del objeto Future devuelto al enviar un mensaje) obviamente aumentará la latencia significativamente (al menos por una red viaje ida y vuelta). Si el cliente usa devoluciones de llamada, la latencia se ocultará, pero el rendimiento estará limitado por la cantidad de mensajes en tránsito (es decir, cuántos mensajes enviará el productor antes de recibir respuestas del servidor).

• Si acks = all, el productor recibirá una respuesta exitosa del broker una vez que todas las réplicas sincronizadas hayan recibido el mensaje. Este es el modo más seguro, ya que puede asegurarse de que más de un corredor tenga el mensaje y de que el mensaje sobrevivirá incluso en el caso de una falla, sin embargo, la latencia que discutimos en el caso acks = 1 será aún mayor, ya que estaremos esperando que más de un broker reciba el mensaje.

buffer.memory

Esto establece la cantidad de memoria que el productor utilizará para almacenar en búfer los mensajes que esperan ser enviados a los intermediarios. Si los mensajes son enviados por la aplicación más rápido de lo que pueden entregarse al servidor, el productor puede quedarse sin espacio y las llamadas send () adicionales bloquearán o lanzarán una excepción, según el parámetro block.on.buffer.full ( reemplazado con max.block.ms en la versión 0.9.0.0, que permite bloquear durante un cierto tiempo y luego lanzar una excepción).

compression.type

De forma predeterminada, los mensajes se envían sin comprimir. Este parámetro se puede establecer en snappy, gzip o lz4, en cuyo caso se utilizarán los algoritmos de compresión correspondientes para comprimir los datos antes de enviarlos a los corredores. Google inventó la compresión rápida para proporcionar relaciones de compresión decentes con una sobrecarga de CPU baja y buena rendimiento, por lo que se recomienda en los casos en que tanto el rendimiento como el ancho de banda sean una preocupación. La compresión Gzip normalmente utilizará más CPU y tiempo, pero dará como resultado mejores relaciones de compresión, por lo que se recomienda en los casos en que el ancho de banda de la red es más restringido. Al habilitar la compresión, reduce la utilización de la red y el almacenamiento  que suele ser un cuello de botella al enviar mensajes a Kafka.

retries

Cuando el productor recibe un mensaje de error del servidor, el error podría ser transitorio (por ejemplo, falta de líder para una partición). En este caso, el valor del parámetro de reintentos controlará cuántas veces el productor reintentará enviar el mensaje antes de darse por vencido y notificar al cliente de un problema. De forma predeterminada, el productor esperará 100 ms entre reintentos, pero puede controlar esto mediante el parámetro retry.backoff.ms. Recomendamos probar cuánto tiempo se tarda en recuperarse de un corredor bloqueado (es decir, cuánto tiempo hasta que todas las particiones obtengan nuevos líderes) y establecer el número de reintentos y el retraso entre ellos de modo que la cantidad total de tiempo dedicado a reintentar sea mayor que el tiempo el clúster de Kafka necesita recuperarse del colapso; de lo contrario, el productor se rendirá demasiado pronto. El productor no volverá a intentar todos los errores. Algunos errores no son transitorios y no provocarán reintentos (p. Ej., Error "mensaje demasiado grande"). En general, debido a que el productor maneja los reintentos por usted, no tiene sentido manejar los reintentos dentro de su propia lógica de aplicación. ustedquerrá centrar sus esfuerzos en el manejo de errores no confiables o casos en los que se agotaron los reintentos.

batch.size

Cuando se envían varios registros a la misma partición, el productor los agrupará. Este parámetro controla la cantidad de memoria en bytes (¡no en mensajes!) que se utilizará para cada lote. Cuando el lote esté lleno, se enviarán todos los mensajes del lote. Sin embargo, esto no significa que el productor esperará a que el lote se llene. El productor enviará lotes medio llenos e incluso lotes con un solo mensaje en ellos. Por lo tanto, establecer un tamaño de lote demasiado grande no causará demoras en el envío de mensajes; solo usará más memoria para los lotes. Establecer el tamaño del lote demasiado pequeño agregará algunos gastos generales porque el productor necesitará enviar mensajes con más frecuencia.

linger.ms

linger.ms controla la cantidad de tiempo para esperar mensajes adicionales antes de enviar el lote actual. KafkaProducer envía un lote de mensajes cuando la corriente el lote de alquiler está lleno o cuando se alcanza el límite de linger.ms. De forma predeterminada, el productor enviará mensajes tan pronto como haya un hilo de remitente disponible para enviarlos, incluso si solo hay un mensaje en el lote. Al establecer linger.ms por encima de 0, le indicamos al productor que espere unos milisegundos para agregar mensajes adicionales al lote antes de enviarlo a los brokers. Esto aumenta la latencia pero también aumenta el rendimiento (debido a que enviamos más mensajes a la vez, hay menos gastos generales por mensaje).

client.id

Esta puede ser cualquier cadena y será utilizada por los intermediarios para identificar los mensajes enviados desde el cliente. Se utiliza en registros y métricas, y para cuotas.

max.in.flight.requests.per.connection

Esto controla cuántos mensajes enviará el productor al servidor sin recibir respuestas. Establecer este valor alto puede aumentar el uso de la memoria mientras mejora rendimiento, pero establecerlo demasiado alto puede reducir el rendimiento a medida que el procesamiento por lotes se vuelve menos eficiente. Establecer esto en 1 garantizará que los mensajes se escribirán al corredor en el orden en que se enviaron, incluso cuando se produzcan reintentos.

timeout.ms, request.timeout.ms, and metadata.fetch.timeout.ms

Estos parámetros controlan cuánto tiempo el productor esperará una respuesta del servidor cuando envíe datos (request.timeout.ms) y cuando solicite metadatos como el líderes actuales para las particiones en las que estamos escribiendo (metadata.fetch.timeout.ms). Si se alcanza el tiempo de espera sin respuesta, el productor volverá a intentar enviar o responder con un error (ya sea a través de una excepción o mediante el envío de devolución de llamada). timeout.ms controla el tiempo que el intermediario esperará para que las réplicas sincronizadas reconozcan el mensaje para cumplir con la configuración de acks; el intermediario devolverá un error si transcurre el tiempo sin los reconocimientos necesarios.

max.block.ms

Este parámetro controla cuánto tiempo bloqueará el productor cuando llame a send () y cuando solicite explícitamente metadatos a través de particionesFor (). Esos métodos bloquean cuando el búfer de envío del productor está lleno o cuando los metadatos no están disponibles. Cuando se alcanza max.block.ms, se lanza una excepción de tiempo de espera.

max.request.size

Esta configuración controla el tamaño de una solicitud de producto enviada por el productor. Limita tanto el tamaño del mensaje más grande que se puede enviar como la cantidad de mensajes que el productor puede enviar en una solicitud. Por ejemplo, con un tamaño de solicitud máximo predeterminado de 1 MB, el mensaje más grande que puede enviar es 1 MB o el productor puede agrupar 1,000 mensajes de tamaño 1 K cada uno en una solicitud. Además, el corredor tiene su propio límite en el tamaño del mensaje más grande que aceptará (message.max.bytes). Por lo general, es una buena idea hacer coincidir estas configuraciones, para que el productor no intente enviar mensajes de un tamaño que el broker rechace.

receive.buffer.bytes y send.buffer.bytes

Estos son los tamaños de los búferes de envío y recepción de TCP que utilizan los sockets al escribir y leer datos. Si se establecen en -1, se utilizarán los valores predeterminados del sistema operativo. Es una buena idea aumentarlos cuando los productores o consumidores se comunican con intermediarios en un centro de datos diferente porque esos enlaces de red suelen tener mayor latencia y menor ancho de banda.

Apache Kafka conserva el orden de los mensajes dentro de una partición. Esto significa que si los mensajes se enviaron desde el productor en un orden específico, el broker los escribirá en una partición en ese orden y todos los consumidores los leerán en ese orden. Para algunos casos de uso, el orden es muy importante. Hay una gran diferencia entre depositar $ 100 en una cuenta y luego retirarlos, ¡y al revés! Sin embargo, algunos casos de uso son menos sensibles.

Establecer el parámetro de reintentos en un valor distinto de cero y el max.in.flights.requests.per.session en más de uno significa que es posible que el bróker no pueda escribir el primer lote de mensajes, logre escribir el segundo (que ya estaba en vuelo), y luego vuelva a intentar el primer lote y tenga éxito, invirtiendo así el orden.

Por lo general, establecer el número de reintentos a cero no es una opción en un sistema confiable, por lo que si garantizar el orden es crítico, recomendamos configurar in.flight.requests.per.session = 1 para asegurarse de que mientras se reintenta un lote de mensajes, no se enviarán mensajes adicionales (porque tiene el potencial de invertir el orden correcto).

Esto limitará severamente el rendimiento del productor, por lo tanto, utilícelo solo cuando el pedido sea importante.

30 años de Linux

El 25 de agosto de 1991, Linus Torvalds presenta el nucleo linux con estas palabras : 

"Hola a todos los que estáis ahí fuera usando minix –

Estoy desarrollando un sistema operativo (libre) (sólo por hobby, no será grande y profesional como gnu) para clónicos de los AT 386(486). Esta idea está madurando desde abril, y ahora está comenzando a estar lista. Me gustaría recibir cualquier comentario en cosas que a la gente le gustan y no le gustan de minix, ya que mi sistema operativo se parece a él un poco (misma disposición física del sistema de ficheros (por razones prácticas) entre otras cosas).

Ya he trasladado bash (1.08) y gcc (1.40) y parece que las cosas funcionan. Esto implica que podría tener algo práctico en pocos meses, y me gustaría saber qué características le gustarían más a la gente tener. Será bienvenida cualquier sugerencia, pero no prometo que las implementaré todas ellas :-) "

Este nucleo nos acompaña desde entonces y hace mucho mejores nuestras vidas....

Feliz Cumpleaños!!! 

Imperativo vs reactivo

Para comprender mejor el contraste entre los 2 mundos, necesitamos explicar la diferencia entre los modelos de ejecución reactiva e imperativa. 

En el enfoque tradicional e imperativo, los frameworks asignan un hilo para manejar una solicitud. Entonces, todo el procesamiento de la solicitud se ejecuta en este hilo de trabajo. Este modelo no escala muy bien. De hecho, para manejar múltiples solicitudes simultáneas, necesita múltiples subprocesos; por lo que la simultaneidad de su aplicación está limitada por el número de subprocesos. Además, estos subprocesos se bloquean tan pronto como su código interactúe con servicios remotos. Por lo tanto, conduce a un uso ineficiente de los recursos, ya que es posible que necesite más subprocesos, y cada subproceso, ya que están asignados a subprocesos del sistema operativo, tiene un costo en términos de memoria y CPU.

Por otro lado, el modelo reactivo se basa en E/S sin bloqueo y en un modelo de ejecución diferente. La E/S sin bloqueo proporciona una forma eficaz de tratar las E/S simultáneas. Una cantidad mínima de subprocesos denominados subprocesos de E/S puede manejar muchas E/S simultáneas. Con un modelo de este tipo, el procesamiento de solicitudes no se delega a un subproceso de trabajo, sino que utiliza estos subprocesos de E/S directamente. Ahorra memoria y CPU, ya que no es necesario crear subprocesos de trabajo para manejar las solicitudes. También mejora la simultaneidad ya que elimina la restricción sobre el número de subprocesos. Finalmente, también mejora el tiempo de respuesta ya que reduce el número de cambios de hilo.

Entonces, con el modelo de ejecución reactiva, las solicitudes se procesan mediante subprocesos de E/S. Pero eso no es todo. Un subproceso de E/S puede manejar varias solicitudes simultáneas. ¿Cómo? Aquí está el truco y una de las diferencias más significativas entre reactivo e imperativo.

Cuando el procesamiento de una solicitud requiere interactuar con un servicio remoto, como una API HTTP o una base de datos, no bloquea la ejecución mientras espera la respuesta. En cambio, programa la operación de E/S y adjunta una continuación, es decir, el código restante de procesamiento de la solicitud. Esta continuación se puede pasar como una devolución de llamada (una función invocada con el resultado de E/S) o utilizar construcciones más avanzadas como programación reactiva o co-rutinas. 

Independientemente de cómo se exprese la programación no bloqueante, lo esencial es la liberación del subproceso de E/S y, en consecuencia, el hecho de que este subproceso pueda utilizarse para procesar otra solicitud. Cuando se completa la E/S, el subproceso de E/S continúa el procesamiento de la solicitud pendiente.

Entonces, a diferencia del modelo imperativo, donde la E/S bloquea la ejecución, los conmutadores reactivos a un diseño basado en la continuación, donde se liberan los subprocesos de E/S y se invoca la continuación cuando se completan las E/S. Como resultado, el subproceso de E/S puede manejar múltiples solicitudes concurrentes, mejorando la concurrencia general de la aplicación.

Pero hay una trampa. Necesitamos una forma de escribir código continuation-passing o no bloqueante. Hay muchas maneras de hacer esto. Por ejemplo en Quarkus propone:

• Mutiny: una biblioteca de programación reactiva intuitiva e impulsada por eventos
• Co-rutinas de Kotlin: una forma de escribir código asincrónico de manera secuencial

Dejo link: https://quarkus.io/guides/getting-started-reactive