DEV Community: @alonso_isidoro

Como conseguir y la consumición entregada y ordenada de mensajes con Kafka

@alonso_isidoro — Mon, 04 Nov 2024 10:45:44 +0000

Para asegurar que los eventos se envían y consumen en un orden totalmente consistente en Apache Kafka, es fundamental entender cómo funciona la partición de mensajes y la asignación de consumidores.

Uso de Particiones en Kafka

Particionamiento de Temas:
- Kafka organiza los mensajes en particiones dentro de un tema. Cada partición mantiene el orden de los mensajes que recibe, lo que significa que los mensajes se procesan en el orden en que fueron enviados a esa partición.
- Para garantizar el orden, es crucial que todos los mensajes relacionados con un mismo contexto (por ejemplo, un ID de usuario o un ID de transacción) se envíen a la misma partición. Esto se logra utilizando una clave de partición al enviar mensajes. Kafka utiliza esta clave para determinar a qué partición enviar el mensaje mediante una función hash[1][5].
Claves de Mensaje:
- Al enviar un mensaje, se puede especificar una clave. Todos los mensajes con la misma clave serán enviados a la misma partición, lo que asegura que sean consumidos en el mismo orden en que fueron producidos. Por ejemplo, si se usa el ID del usuario como clave, todos los eventos relacionados con ese usuario irán a la misma partición.

Grupos de Consumidores

Asignación de Consumidores:
- Los consumidores en Kafka se agrupan en grupos de consumidores. Cada grupo puede tener múltiples consumidores, pero cada partición solo puede ser leída por un consumidor dentro del grupo a la vez.
- Esto significa que si se tienen más consumidores que particiones, algunos consumidores estarán inactivos. Para mantener el orden y maximizar la eficiencia, es recomendable tener al menos tantas particiones como consumidores en el grupo.
Manejo de Offsets:
- Kafka almacena el estado de lectura de cada consumidor mediante offsets, que son identificadores numéricos incrementales para cada mensaje dentro de una partición. Esto permite que los consumidores retomen la lectura desde donde lo dejaron en caso de fallos.

Estrategias Adicionales

Evitar Sobrecargas: Al elegir claves de partición, es importante considerar la distribución del tráfico para evitar que algunas particiones estén sobrecargadas mientras otras están subutilizadas.
Replicación y Tolerancia a Fallos: Asegúrate de configurar una replicación adecuada (mayor a 1) para las particiones, lo que no solo mejora la disponibilidad sino también la resiliencia del sistema ante fallos.

Para implementar un sistema de producción y consumo de mensajes en Kafka utilizando Avro, asegurando que los mensajes se procesen en orden y manejando posibles fallos, aquí tienes un ejemplo completo. Este incluye la definición del esquema Avro, el código del productor y consumidor, así como estrategias para manejar errores.
Esquema Avro
Primero, definimos el esquema Avro para nuestro payload. Crearemos un archivo llamado user_signed_up.avsc que describe la estructura del mensaje.

{
  "type": "record",
  "name": "UserSignedUp",
  "namespace": "com.example",
  "fields": [
    { "name": "userId", "type": "int" },
    { "name": "userEmail", "type": "string" },
    { "name": "timestamp", "type": "string" } // Formato ISO 8601
  ]
}

Generación de la Clave
Para asegurar el orden en la producción y consumo de mensajes, utilizaremos una clave estructurada como tipo-mensaje-fecha, por ejemplo: user-signed-up-2024-11-04.
Productor Kafka
Aquí tienes el código del productor que envía mensajes a Kafka utilizando el esquema Avro:

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.util.Properties;

public class AvroProducer {
    private final KafkaProducer<String, byte[]> producer;
    private final Schema schema;

    public AvroProducer(String bootstrapServers) throws IOException {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "io.confluent.kafka.serializers.KafkaAvroSerializer");

// Establecer la propiedad de reintentos, Número de reintentos
        properties.put(ProducerConfig.RETRIES_CONFIG, 3); 
// Asegura que todos los réplicas reconozcan la escritura,
        properties.put(ProducerConfig.ACKS_CONFIG, "all"); 
// Solo un mensaje a la vez
properties.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1); 
// Habilitar idempotencia, no quiero enviar duplicados
properties.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); 

        this.producer = new KafkaProducer<>(properties);
        this.schema = new Schema.Parser().parse(new File("src/main/avro/user_signed_up.avsc"));
    }

    public void sendMessage(String topic, int userId, String userEmail) {
        GenericRecord record = new GenericData.Record(schema);
        record.put("userId", userId);
        record.put("userEmail", userEmail);
        record.put("timestamp", java.time.Instant.now().toString());

        String key = String.format("user-signed-up-%s", java.time.LocalDate.now());

        ProducerRecord<String, byte[]> producerRecord = new ProducerRecord<>(topic, key, serialize(record));

        producer.send(producerRecord, (metadata, exception) -> {
            if (exception != null) {
                exception.printStackTrace();
**handleFailure(exception, producerRecord);
**            } else {
                System.out.printf("Mensaje enviado a la partición %d con offset %d%n", metadata.partition(), metadata.offset());
            }
        });
    }

private void handleFailure(Exception exception, ProducerRecord<String, byte[]> producerRecord) {
        // Log the error for monitoring
        System.err.println("Error sending message: " + exception.getMessage());

        // Implement local persistence as a fallback
        saveToLocalStorage(producerRecord);

        // Optionally: Notify an external monitoring system or alert
    }

    private void saveToLocalStorage(ProducerRecord<String, byte[]> record) {
        try {
            // Persist the failed message to a local file or database for later processing
            Files.write(new File("failed_messages.log").toPath(), 
                         (record.key() + ": " + new String(record.value()) + "\n").getBytes(), 
                         StandardOpenOption.CREATE, 
                         StandardOpenOption.APPEND);
            System.out.println("Mensaje guardado localmente para reenvío: " + record.key());
        } catch (IOException e) {
            System.err.println("Error saving to local storage: " + e.getMessage());
        }
    }

    private byte[] serialize(GenericRecord record) {
        // Crear un ByteArrayOutputStream para almacenar los bytes serializados
    ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
    // Crear un escritor de datos para el registro Avro
    DatumWriter<GenericRecord> datumWriter = new GenericDatumWriter<>(record.getSchema());

    // Crear un encoder para escribir en el ByteArrayOutputStream
    Encoder encoder = EncoderFactory.get().binaryEncoder(outputStream, null);

    try {
        // Escribir el registro en el encoder
        datumWriter.write(record, encoder);
        // Finalizar la escritura
        encoder.flush();
    } catch (IOException e) {
        throw new AvroSerializationException("Error serializing Avro record", e);
    }

    // Devolver los bytes serializados
    return outputStream.toByteArray();
    }

    public void close() {
        producer.close();
    }
}

Consideraciones sobre Reintentos
**Es importante tener en cuenta que al habilitar reintentos, puede haber un riesgo de reordenamiento de mensajes si no se maneja adecuadamente.
Para evitar esto:
**max.in.flight.requests.per.connection: Puedes establecer esta propiedad en 1 para garantizar que los mensajes se envíen uno a la vez y se procesen en orden. Sin embargo, esto puede afectar el rendimiento.
Con esta configuración y manejo adecuado de errores, puedes asegurar que tu productor Kafka sea más robusto y capaz de manejar fallos en la producción de mensajes mientras mantienes el orden necesario.

**Consumidor Kafka
**El consumidor que lee y procesa los mensajes:

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.io.DatumReader;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.io.File;
import java.io.IOException;
import java.util.Collections;
import java.util.Properties;

public class AvroConsumer {
    private final KafkaConsumer<String, byte[]> consumer;
    private final Schema schema;

    public AvroConsumer(String bootstrapServers, String groupId) throws IOException {
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "io.confluent.kafka.serializers.KafkaAvroDeserializer");

        this.consumer = new KafkaConsumer<>(properties);
        this.schema = new Schema.Parser().parse(new File("src/main/avro/user_signed_up.avsc"));
    }

    public void consume(String topic) {
        consumer.subscribe(Collections.singletonList(topic));

        while (true) {
            ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, byte[]> record : records) {
                try {
                    processMessage(record.value());
                } catch (Exception e) {
                    handleProcessingError(e, record);
                }
            }
        }
    }

    private void processMessage(byte[] data) throws IOException {
        DatumReader<GenericRecord> reader = new GenericDatumReader<>(schema);
        var decoder = DecoderFactory.get().binaryDecoder(data, null);
        GenericRecord record = reader.read(null, decoder);

        System.out.printf("Consumido mensaje: %s - %s - %s%n", 
            record.get("userId"), 
            record.get("userEmail"), 
            record.get("timestamp"));
    }

    private void handleProcessingError(Exception e, ConsumerRecord<String, byte[]> record) {
        System.err.println("Error processing message: " + e.getMessage());

        // Implement logic to save failed messages for later processing
        saveFailedMessage(record);
    }

    private void saveFailedMessage(ConsumerRecord<String, byte[]> record) {
        try {
            // Persist the failed message to a local file or database for later processing
            Files.write(new File("failed_consumed_messages.log").toPath(), 
                         (record.key() + ": " + new String(record.value()) + "\n").getBytes(), 
                         StandardOpenOption.CREATE,
                         StandardOpenOption.APPEND);
            System.out.println("Mensaje consumido guardado localmente para re-procesamiento: " + record.key());
        } catch (IOException e) {
            System.err.println("Error saving consumed message to local storage: " + e.getMessage());
        }
    }

    public void close() {
        consumer.close();
    }
}

Ejemplo Realista de Claves
En un entorno con muchos eventos distintos y muchas particiones distintas, una clave realista podría ser algo como:

user-signed-up-2024-11-04
order-created-2024-11-04
payment-processed-2024-11-04

Esto permite que todos los eventos relacionados con un tipo específico en una fecha específica se envíen a la misma partición y se procesen en orden. Además, puedes diversificar las claves incluyendo más detalles si es necesario (como un ID de sesión o transacción).
Con esta implementación y estrategias para manejar fallos y asegurar el orden de los mensajes en Kafka utilizando Avro, puedes construir un sistema robusto y eficiente para gestionar eventos.

Ahora un Productor y un consumidor Kafka algo más capaz.

Productor Kafka con Circuit Breaker, Persistencia Local y DLQ.

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import io.github.resilience4j.circuitbreaker.CircuitBreaker;
import io.github.resilience4j.circuitbreaker.CircuitBreakerConfig;
import io.github.resilience4j.circuitbreaker.CircuitBreakerRegistry;

import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.time.Duration;
import java.util.Properties;

public class AvroProducer {
    private final KafkaProducer<String, byte[]> producer;
    private final Schema schema;
    private final CircuitBreaker circuitBreaker;

    public AvroProducer(String bootstrapServers) throws IOException {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "io.confluent.kafka.serializers.KafkaAvroSerializer");

        properties.put(ProducerConfig.RETRIES_CONFIG, 3);
        properties.put(ProducerConfig.ACKS_CONFIG, "all");
        properties.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1);

        this.producer = new KafkaProducer<>(properties);
        this.schema = new Schema.Parser().parse(new File("src/main/avro/user_signed_up.avsc"));

        // Configuración del Circuit Breaker
        CircuitBreakerConfig config = CircuitBreakerConfig.custom()
                .failureRateThreshold(50) // porcentaje de fallos para abrir el circuito
                .waitDurationInOpenState(Duration.ofSeconds(30)) // tiempo en estado abierto
                .slidingWindowSize(10) // número de llamadas para calcular el porcentaje de fallos
                .build();

        CircuitBreakerRegistry registry = CircuitBreakerRegistry.of(config);
        this.circuitBreaker = registry.circuitBreaker("producerCircuitBreaker");
    }

    public void sendMessage(String topic, int userId, String userEmail) {
        GenericRecord record = new GenericData.Record(schema);
        record.put("userId", userId);
        record.put("userEmail", userEmail);
        record.put("timestamp", java.time.Instant.now().toString());

        String key = String.format("user-signed-up-%s", java.time.LocalDate.now());

        ProducerRecord<String, byte[]> producerRecord = new ProducerRecord<>(topic, key, serialize(record));

        circuitBreaker.executeRunnable(() -> {
            producer.send(producerRecord, (metadata, exception) -> {
                if (exception != null) {
                    handleFailure(exception, producerRecord);
                } else {
                    System.out.printf("Mensaje enviado a la partición %d con offset %d%n", metadata.partition(), metadata.offset());
                }
            });
        });
    }

    private byte[] serialize(GenericRecord record) throws AvroSerializationException {
        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
        DatumWriter<GenericRecord> datumWriter = new GenericDatumWriter<>(record.getSchema());
        Encoder encoder = EncoderFactory.get().binaryEncoder(outputStream, null);

        try {
            datumWriter.write(record, encoder);
            encoder.flush();
        } catch (IOException e) {
            throw new AvroSerializationException("Error serializing Avro record", e);
        }

        return outputStream.toByteArray();
    }

    private void handleFailure(Exception exception, ProducerRecord<String, byte[]> producerRecord) {
        System.err.println("Error al enviar el mensaje: " + exception.getMessage());

        // Guardar el mensaje en una cola de errores (DLQ)
        saveToDeadLetterQueue(producerRecord);

        // Guardar en almacenamiento local como respaldo
        saveToLocalStorage(producerRecord);
    }

    private void saveToDeadLetterQueue(ProducerRecord<String, byte[]> record) {
        try {
            Files.write(new File("dead_letter_queue.log").toPath(), 
                         (record.key() + ": " + new String(record.value()) + "\n").getBytes(), 
                         StandardOpenOption.CREATE,
                         StandardOpenOption.APPEND);
            System.out.println("Mensaje guardado en Dead Letter Queue: " + record.key());
        } catch (IOException e) {
            System.err.println("Error guardando en DLQ: " + e.getMessage());
        }
    }

    private void saveToLocalStorage(ProducerRecord<String, byte[]> record) {
        try {
            Files.write(new File("failed_messages.log").toPath(), 
                         (record.key() + ": " + new String(record.value()) + "\n").getBytes(), 
                         StandardOpenOption.CREATE,
                         StandardOpenOption.APPEND);
            System.out.println("Mensaje guardado localmente para reenvío: " + record.key());
        } catch (IOException e) {
            System.err.println("Error guardando en almacenamiento local: " + e.getMessage());
        }
    }

    public void close() {
        producer.close();
    }
}

Consumidor Kafka con Manejo de DLQ.

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.io.DatumReader;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.util.Collections;
import java.util.Properties;

public class AvroConsumer {
    private final KafkaConsumer<String, byte[]> consumer;
    private final Schema schema;

    public AvroConsumer(String bootstrapServers, String groupId) throws IOException {
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "io.confluent.kafka.serializers.KafkaAvroDeserializer");

        this.consumer = new KafkaConsumer<>(properties);
        this.schema = new Schema.Parser().parse(new File("src/main/avro/user_signed_up.avsc"));
    }

    public void consume(String topic) {
        consumer.subscribe(Collections.singletonList(topic));

        while (true) {
            ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, byte[]> record : records) {
                try {
                    processMessage(record.value());
                } catch (Exception e) {
                    handleProcessingError(e, record);
                }
            }
            processDeadLetterQueue(); // Procesar mensajes en la DLQ después de consumir del tópico principal
        }
    }

    private void processMessage(byte[] data) throws IOException {
        DatumReader<GenericRecord> reader = new GenericDatumReader<>(schema);
        var decoder = DecoderFactory.get().binaryDecoder(data, null);
        GenericRecord record = reader.read(null, decoder);

        System.out.printf("Consumido mensaje: %s - %s - %s%n", 
            record.get("userId"), 
            record.get("userEmail"), 
            record.get("timestamp"));
    }

    private void handleProcessingError(Exception e, ConsumerRecord<String, byte[]> record) {
        System.err.println("Error procesando el mensaje: " + e.getMessage());

        // Guardar el mensaje fallido en la DLQ
        saveFailedMessage(record);
    }

    private void saveFailedMessage(ConsumerRecord<String, byte[]> record) {
        try {
            Files.write(new File("dead_letter_queue.log").toPath(), 
                         (record.key() + ": " + new String(record.value()) + "\n").getBytes(), 
                         StandardOpenOption.CREATE,
                         StandardOpenOption.APPEND);
            System.out.println("Mensaje consumido guardado en Dead Letter Queue: " + record.key());

            // Opcionalmente puedes eliminar el mensaje del tópico original si es necesario
            // consumer.commitSync(); // Confirmar el procesamiento si es necesario

         } catch (IOException e) {
             System.err.println("Error guardando mensaje consumido en DLQ: " + e.getMessage());
         }
     }

     private void processDeadLetterQueue() {
         try {
             Files.lines(new File("dead_letter_queue.log").toPath()).forEach(line -> {
                 System.out.println("Procesando mensaje desde la Dead Letter Queue: " + line);
                 // Aquí puedes implementar la lógica para reintentar o procesar mensajes desde la DLQ.
             });
         } catch (IOException e) {
             System.err.println("Error procesando la Dead Letter Queue: " + e.getMessage());
         }
     }

     public void close() {
         consumer.close();
     }
}

<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.example</groupId>
    <artifactId>kafka-avro-example</artifactId>
    <version>1.0-SNAPSHOT</version>
    <packaging>jar</packaging>

    <properties>
        <java.version>17</java.version>
        <kafka.version>3.5.1</kafka.version>
        <avro.version>1.11.0</avro.version>
        <resilience4j.version>1.7.1</resilience4j.version>
        <slf4j.version>1.7.36</slf4j.version>
    </properties>

    <dependencies>
        <!-- Kafka Clients -->
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>${kafka.version}</version>
        </dependency>

        <!-- Avro -->
        <dependency>
            <groupId>org.apache.avro</groupId>
            <artifactId>avro</artifactId>
            <version>${avro.version}</version>
        </dependency>

        <!-- Confluent Kafka Avro Serializer -->
        <dependency>
            <groupId>io.confluent</groupId>
            <artifactId>kafka-avro-serializer</artifactId>
            <version>7.5.0</version> <!-- Asegúrate de que esta versión sea compatible con tu versión de Kafka -->
        </dependency>

        <!-- Resilience4j -->
        <dependency>
            <groupId>io.github.resilience4j</groupId>
            <artifactId>resilience4j-spring-boot2</artifactId>
            <version>${resilience4j.version}</version>
        </dependency>

        <!-- SLF4J API -->
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>${slf4j.version}</version>
        </dependency>

        <!-- Logging Implementation (Logback) -->
        <dependency>
            <groupId>ch.qos.logback</groupId>
            <artifactId>logback-classic</artifactId>
            <version>1.2.6</version>
        </dependency>

        <!-- JUnit for Testing -->
        <dependency>
            <groupId>org.junit.jupiter</groupId>
            <artifactId>junit-jupiter-engine</artifactId>
            <version>5.8.2</version>
            <scope>test</scope>
        </dependency>

    </dependencies>

    <build>
        <plugins>

            <!-- Maven Compiler Plugin -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.11.0</version>
                <configuration>
                    <source>${java.version}</source>
                    <target>${java.version}</target>
                </configuration>
            </plugin>

        </plugins>
    </build>

</project>

Explicación del Código
Circuit Breaker:
Se utiliza Resilience4j para gestionar el circuito breaker del productor. Se configura un umbral de tasa de fallo y un tiempo de espera en estado abierto.
Persistencia Local y DLQ:
Los mensajes fallidos se guardan tanto en un archivo local (failed_messages.log) como en una cola de errores (dead_letter_queue.log).
Manejo de Errores:
En el productor y consumidor se manejan los errores adecuadamente y se registran.
Procesamiento de la DLQ:
El consumidor también procesa los mensajes almacenados en la DLQ después de consumir mensajes del tópico principal.
Logging:
Se utilizan mensajes System.err y System.out para registrar errores y eventos importantes.
Consideraciones Finales
Con esta implementación:
Se asegura que los mensajes sean enviados y procesados de manera resiliente.
Se proporciona un manejo adecuado para errores temporales o persistentes.
La lógica permite una recuperación efectiva mediante el uso de una Dead Letter Queue.
El circuito breaker ayuda a evitar que el sistema se sature ante fallos prolongados.
Este enfoque crea un sistema robusto que puede manejar problemas físicos y lógicos mientras mantiene la entrega ordenada de mensajes en Kafka.

IndexIVFFlat y IndexIVFPQ

@alonso_isidoro — Wed, 09 Oct 2024 10:45:12 +0000

Here's a comparison between the IndexIVFFlat and IndexIVFPQ indices, along with some alternatives for their use:

Comparison: IndexIVFFlat vs. IndexIVFPQ

Characteristic	IndexIVFFlat	IndexIVFPQ
Storage Type	Stores vectors in their original form.	Utilizes product quantization (PQ) to compress vectors.
Precision	High precision, as it performs exact searches within cells.	May sacrifice some precision for compression, but still provides good results.
Search Speed	Slower on large datasets due to exhaustive search.	Faster, especially on large sets, thanks to reduced search space.
Memory Usage	Consumes more memory as it stores all vectors without compression.	Consumes significantly less memory due to compression (up to 97% less).
Configuration	Simpler, only requires defining the number of cells (nlist).	Requires defining both the number of cells (nlist) and code size (code_size).
Training	Needs to be trained to create cells before adding data.	Also requires training, but the process is more complex due to quantization.

Pros and Cons

Pros of IndexIVFFlat

Precision: Provides exact results when searching within each cell.
Simplicity: Easy to understand and configure.

Cons of IndexIVFFlat

Speed: Can be very slow with large volumes of data.
Memory Usage: Does not optimize memory usage, which can be problematic with large datasets.

Pros of IndexIVFPQ

Speed: Much faster in searches due to reduced search space.
Memory Efficiency: Significantly reduces memory usage, allowing for handling larger datasets.

Cons of IndexIVFPQ

Precision: There may be a slight loss in precision due to compression.
Complexity: Configuration and training are more complex than in IndexIVFFlat.

Alternatives

IndexFlatL2
- Performs an exhaustive search without compression. Ideal for small datasets where maximum precision is required.
IndexPQ
- Uses only product quantization without clustering. Useful when a balance between speed and precision is needed, but clustering is not required.
IndexIVFScalarQuantizer
- Combines the inverted index with scalar quantization, offering a different approach to reduce memory usage and improve speed.
IndexIVFPQR
- A variant that combines IVF and PQ with code-based re-ranking, offering a balance between speed and improved precision.
Composite Indexes
- Use index_factory to create composite indices that combine multiple techniques (e.g., OPQ + IVF + PQ) to further optimize performance.

These alternatives allow adapting the solution to different needs in terms of precision, speed, and memory usage according to the specific case being addressed.

Citations:
[1] https://github.com/facebookresearch/faiss/wiki/Faiss-indexes/9df19586b3a75e4cb1c2fb915f2c695755a599b8
[2] https://ai.plainenglish.io/speeding-up-similarity-search-in-recommender-systems-with-faiss-advanced-concepts-part-ii-95e796a7db74?gi=ce57aff1a0c4
[3] https://www.pinecone.io/learn/series/faiss/faiss-tutorial/
[4] https://faiss.ai/cpp_api/struct/structfaiss_1_1IndexIVFFlat.html
[5] https://unfoldai.com/effortless-large-scale-image-retrieval-with-faiss-a-hands-on-tutorial/
[6] https://www.pinecone.io/learn/series/faiss/product-quantization/
[7] https://www.pinecone.io/learn/series/faiss/composite-indexes/
[8] https://github.com/facebookresearch/faiss/issues/1113

En Español, Soy Español, pero por respeto a la comunidad, pongo primero la traduccion al inglés.

Aquí tienes una comparación entre los índices IndexIVFFlat e IndexIVFPQ, junto con algunas alternativas para su uso:

Comparación: IndexIVFFlat vs. IndexIVFPQ

Característica	IndexIVFFlat	IndexIVFPQ
Tipo de Almacenamiento	Almacena vectores en su forma original.	Utiliza cuantización de producto (PQ) para comprimir vectores.
Precisión	Alta precisión, ya que realiza búsquedas exactas dentro de las celdas.	Puede sacrificar algo de precisión por la compresión, pero aún proporciona buenos resultados.
Velocidad de Búsqueda	Más lento en grandes conjuntos de datos debido a la búsqueda exhaustiva.	Más rápido, especialmente en grandes conjuntos, gracias a la reducción del espacio de búsqueda.
Uso de Memoria	Consume más memoria porque almacena todos los vectores sin compresión.	Consume significativamente menos memoria debido a la compresión (hasta 97% menos).
Configuración	Más simple, solo requiere definir el número de celdas (nlist).	Requiere definir tanto el número de celdas (nlist) como el tamaño del código (code_size).
Entrenamiento	Necesita ser entrenado para crear las celdas antes de añadir datos.	También necesita entrenamiento, pero el proceso es más complejo debido a la cuantización.

Pros y Contras

Pros de IndexIVFFlat

Precisión: Proporciona resultados exactos al buscar en cada celda.
Simplicidad: Fácil de entender y configurar.

Contras de IndexIVFFlat

Velocidad: Puede ser muy lento con grandes volúmenes de datos.
Uso de Memoria: No optimiza el uso de memoria, lo que puede ser un problema con conjuntos de datos grandes.

Pros de IndexIVFPQ

Velocidad: Mucho más rápido en búsquedas debido a la reducción del espacio de búsqueda.
Eficiencia de Memoria: Reduce significativamente el uso de memoria, lo que permite manejar conjuntos de datos más grandes.

Contras de IndexIVFPQ

Precisión: Puede haber una ligera pérdida en la precisión debido a la compresión.
Complejidad: La configuración y entrenamiento son más complejos que en IndexIVFFlat.

Alternativas

IndexFlatL2
- Realiza una búsqueda exhaustiva sin compresión. Ideal para conjuntos de datos pequeños donde se requiere precisión máxima.
IndexPQ
- Utiliza solo cuantización de producto sin agrupación. Es útil cuando se necesita un balance entre velocidad y precisión, pero no se requiere agrupación.
IndexIVFScalarQuantizer
- Combina el índice invertido con cuantización escalar, ofreciendo un enfoque diferente para reducir el uso de memoria y mejorar la velocidad.
IndexIVFPQR
- Una variante que combina IVF y PQ con re-ranking basado en códigos, ofreciendo un equilibrio entre velocidad y precisión mejorada.
Composite Indexes
- Usar index_factory para crear índices compuestos que combinan múltiples técnicas (por ejemplo, OPQ + IVF + PQ) para optimizar aún más el rendimiento.

Estas alternativas permiten adaptar la solución a diferentes necesidades en cuanto a precisión, velocidad y uso de memoria según el caso específico que estés abordando.

Citations:
[1] https://www.pinecone.io/learn/series/faiss/faiss-tutorial/
[2] https://www.pinecone.io/learn/series/faiss/product-quantization/
[3] https://www.pinecone.io/learn/series/faiss/composite-indexes/
[4] https://github.com/facebookresearch/faiss/wiki/Faiss-indexes/9df19586b3a75e4cb1c2fb915f2c695755a599b8
[5] https://faiss.ai/cpp_api/struct/structfaiss_1_1IndexIVFFlat.html
[6] https://pub.towardsai.net/unlocking-the-power-of-efficient-vector-search-in-rag-applications-c2e3a0c551d5?gi=71a82e3ea10e
[7] https://www.pingcap.com/article/mastering-faiss-vector-database-a-beginners-handbook/
[8] https://wangzwhu.github.io/home/file/acmmm-t-part3-ann.pdf
[9] https://github.com/alonsoir/ubiquitous-carnival/blob/main/contextual-data-faiss-IndexIVFPQ.py
[10] https://github.com/alonsoir/ubiquitous-carnival/blob/main/contextual-data-faiss-indexivfflat.py

Oracle-JDK23

@alonso_isidoro — Thu, 03 Oct 2024 10:41:56 +0000

*JDK 23-Oracle.txt
*
Esto es una sesión con la jshell después de instalar el jdk 23-oracle con sdkman.

jshell --enable-preview
|  Welcome to JShell -- Version 23
|  For an introduction type: /help intro

La JDK 23 nos trae varias novedades:

1. Tipos Primitivos en Patterns (JEP 455)

2. Tipos Primitivos en Patterns (JEP 476)

3. Scoped Values (JEP 481)

4. Clases y Métodos de Instancia Declarados Implícitamente (JEP 477)

5. Class-File API (JEP 466)

6. Cuerpos de Constructor Flexibles (JEP 482)

7. Comentarios en Markdown para Documentación (JEP 467)

*1. Tipos Primitivos en Patterns (JEP 455)
*

Veamoslo en un ejemplo.

jshell> 
System.out.println("Pattern Matching New Model.... Primitive Supported");
       Object o = 127;
        switch (o) {
            case int i -> System.out.println("Integer: " + i);
            case long l -> System.out.println("Long: " + l);
            default -> System.out.println("Other: " + o);
        }
Pattern Matching New Model.... Primitive Supported
o ==> 127
Integer: 127

*2. Tipos Primitivos en Patterns (JEP 476)
*
jshell>
// New way of Java
import static java.lang.System.out;
import module java.base;

    public class _3_ModuleImport {

        public static void main (String[] args) throws Exception{
            // Multiple API's are called with a single import.  JEP 476
            // import module java.base
            System.out.println("java --enable-preview _3_ModuleImport.java");
            System.out.println("Array of Values  = "+List.of("Hello ", "Module Import ", "World!"));
            File file = new File("README.MD");
            System.out.println("File README.MD = "+file.canRead());
            System.out.println("Reading the file....");
            try (BufferedReader br = new BufferedReader(new FileReader(file))) {
                while(br.ready()) {
                    System.out.println(br.readLine());
                }
            }
            System.out.println("File closed ....");

        }
    }

Me aseguro que en el directorio donde he generado .java anterior tiene un README.md

    ls README.md
    README.md

Compilo y ejecuto el nuevo fichero, ojito que hay que activar los flags.

    javac --enable-preview --release 23 _3_ModuleImport.java
    java --enable-preview _3_ModuleImport
    java --enable-preview _3_ModuleImport.java

Por lo visto, da igual usar el .java que no usarlo. Demasiado tiempo con python, supongo...

    Array of Values = [Hello , Module Import , World!]
    File README.MD = true
    Reading the file....
    hola esto es un fichero README de pruebas para que spark-shell-3.5.0 pueda leerlo.
    File closed ....

*3. Scoped Values (JEP 481)
*
Concepto principal

Un "scoped value" es un objeto contenedor que permite compartir un valor de datos inmutable entre un método y sus llamadas directas e indirectas dentro del mismo hilo, así como con hilos secundarios2. Se declara típicamente como un campo estático final de tipo ScopedValue.

Ventajas sobre variables thread-local
    Los Scoped Values ofrecen varias ventajas sobre las variables thread-local tradicionales:

    Mayor facilidad de razonamiento sobre el flujo de datos

    Menor costo en espacio y tiempo, especialmente al usarse con hilos virtuales y concurrencia estructurada

    Garantía de inmutabilidad de los datos compartidos

    Ciclo de vida más predecible y acotado

Ejemplo: Sistema de Comercio Electrónico

Configuración de Scoped Values

    public class ApplicationContext {
        public static final ScopedValue<UserSession> USER_SESSION = ScopedValue.newInstance();
        public static final ScopedValue<AuditInfo> AUDIT_INFO = ScopedValue.newInstance();
    }

Filtro de Servlet para inicializar el contexto

    @WebFilter("/*")
    public class ContextInitializationFilter implements Filter {
        @Override
        public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) 
                throws IOException, ServletException {
            HttpServletRequest httpRequest = (HttpServletRequest) request;
            UserSession userSession = authenticateAndCreateUserSession(httpRequest);
            AuditInfo auditInfo = new AuditInfo(UUID.randomUUID().toString());

            ScopedValue.where(ApplicationContext.USER_SESSION, userSession)
                .where(ApplicationContext.AUDIT_INFO, auditInfo)
                .run(() -> chain.doFilter(request, response));
        }

        private UserSession authenticateAndCreateUserSession(HttpServletRequest request) {
            // Lógica de autenticación
            // ...
        }
    }

Servicio de Carrito de Compras

    @Service
    public class ShoppingCartService {
        @Autowired
        private ProductRepository productRepository;
        @Autowired
        private CartRepository cartRepository;

        public void addToCart(String productId, int quantity) {
            UserSession userSession = ApplicationContext.USER_SESSION.get();
            AuditInfo auditInfo = ApplicationContext.AUDIT_INFO.get();

            Product product = productRepository.findById(productId)
                .orElseThrow(() -> new ProductNotFoundException(productId));

            Cart cart = cartRepository.findByUserId(userSession.getUserId())
                .orElseGet(() -> new Cart(userSession.getUserId()));

            cart.addItem(new CartItem(product, quantity));
            cartRepository.save(cart);

            logAuditEvent(auditInfo, "ADD_TO_CART", Map.of(
                "userId", userSession.getUserId(),
                "productId", productId,
                "quantity", quantity
            ));
        }

        private void logAuditEvent(AuditInfo auditInfo, String eventType, Map<String, Object> details) {
            // Lógica para registrar el evento de auditoría
            // ...
        }
    }

Controlador REST

    @RestController
    @RequestMapping("/api/cart")
    public class ShoppingCartController {
        @Autowired
        private ShoppingCartService shoppingCartService;

        @PostMapping("/add")
        public ResponseEntity<String> addToCart(@RequestBody AddToCartRequest request) {
            shoppingCartService.addToCart(request.getProductId(), request.getQuantity());
            return ResponseEntity.ok("Producto añadido al carrito");
        }
    }

** Explicación del ejemplo
**
Configuración: Definimos dos ScopedValue: uno para la sesión del usuario y otro para la información de auditoría.

Filtro de Servlet: Inicializa los ScopedValue para cada solicitud HTTP. Esto asegura que la información de sesión y auditoría esté disponible para todo el procesamiento de la solicitud.

Servicio de Carrito: Utiliza los ScopedValue para acceder a la información de sesión del usuario y los detalles de auditoría sin necesidad de pasarlos explícitamente como parámetros.

Controlador: Simplemente llama al servicio, sin preocuparse por pasar información de contexto.

Beneficios en este escenario

Limpieza del código: Los métodos no necesitan pasar explícitamente la información de sesión o auditoría.

Seguridad: La información de sesión está contenida y no puede ser modificada accidentalmente.

Facilidad de testing: Es más fácil simular diferentes contextos en pruebas unitarias.

Rendimiento: Especialmente útil con hilos virtuales, ya que no hay sobrecarga de almacenamiento por hilo.

Este ejemplo muestra cómo los Scoped Values pueden simplificar significativamente el manejo de información contextual en una aplicación web, mejorando la legibilidad del código y la seguridad de los datos.

*4. Clases y Métodos de Instancia Declarados Implícitamente (JEP 477)
*
Esta característica, que se encuentra en su tercera versión preliminar en el JDK 23, está diseñada para hacer que Java sea más accesible para los principiantes y simplificar el código para todos los desarrolladores5.

** Objetivo principal
**
El propósito es permitir declaraciones de clase más simples y una expansión más fácil de los programas, especialmente para los nuevos programadores, un tanto más vagos que las anteriores generaciones. En mi opinión, han introducido esto para atraer a los programadores python que se atragantan con tanta verbosidad. :)

** Características clave
**
Declaraciones de clase simplificadas: Permite omitir la declaración explícita de la clase en programas simples.

Métodos de instancia main: Introduce la posibilidad de tener un método main de instancia en lugar del tradicional método estático.

Importaciones automáticas: El JDK 23 incluye la importación automática de tres métodos estáticos para E/S de texto y todas las clases e interfaces públicas de nivel superior de los paquetes exportados por el módulo java.base5.

** Ejemplo de uso
** Antes de esta característica, un programa simple en Java requería una estructura como esta:

    public class HelloWorld {
        public static void main(String[] args) {
            System.out.println("Hello, World!");
        }
    }

Con las clases y métodos de instancia declarados implícitamente, el mismo programa podría escribirse así:

    void main() {
        System.out.println("Hello, World!");
    }

**Beneficios**

**Reducción de la verbosidad**: Elimina la necesidad de declarar explícitamente la clase y el método estático main para programas simples.

**Facilita el aprendizaje**: Hace que Java sea más accesible para principiantes, permitiéndoles centrarse en la lógica del programa sin preocuparse por la estructura de clases en etapas iniciales.

**Evolución gradual**: Permite a los programadores comenzar con programas simples y expandirlos gradualmente a medida que crecen en complejidad.

**Alineación con otros lenguajes**: Se asemeja a la estructura de programas en lenguajes como Python o JavaScript, lo que puede facilitar la transición a Java para desarrolladores de otros lenguajes5.

**Contexto y comparación**
Esta característica se inspira en enfoques similares adoptados por otros lenguajes y frameworks populares entre los principiantes, como Node.js. C# implementó una funcionalidad parecida hace algunas versiones con el mismo objetivo de simplificar la entrada al lenguaje5.

**Conclusión**
Las Clases y Métodos de Instancia Declarados Implícitamente representan un esfuerzo significativo por parte de Java para modernizar su sintaxis y hacerla más accesible, especialmente para nuevos programadores. Aunque está en su tercera versión preliminar, esta característica promete cambiar la forma en que se escriben los programas Java simples, allanando el camino para una curva de aprendizaje más suave y una transición más fácil desde otros lenguajes de programación.

Class-File API (JEP 466)

** Propósito y funcionalidad
** La Class-File API está diseñada para parsear, generar y transformar archivos de clase Java (.class). Se define en el paquete java.lang.classfile y proporciona una interfaz estándar para procesar archivos de clase que se alinea con el formato definido en la Especificación de la Máquina Virtual de Java12.

** Características principales
**
Representación inmutable: Todos los elementos del archivo de clase (campos, métodos, atributos, instrucciones de bytecode, etc.) se representan como objetos inmutables12.

**Estructura de árbol**: Refleja la naturaleza jerárquica de los archivos de clase2.

**Navegación dirigida por el usuario**: Permite una análisis eficiente1.

**Parseo perezoso**: Procesa solo las partes del archivo de clase que el usuario requiere1.

**Transformación como propiedad emergente**: No requiere un modo especial o una superficie de API significativamente nueva para las transformaciones1.

** Abstracciones principales
**

Elementos: Descripciones inmutables de componentes del archivo de clase.
Constructores (Builders): Facilitan la construcción de archivos de clase.
Transformaciones: Funciones que modifican elementos durante el proceso de construcción2.

**Beneficios**

**Evolución conjunta**: La API evolucionará junto con el formato de archivo de clase, facilitando la adopción rápida de nuevas características del lenguaje y la JVM24


**Estandarización**: Proporciona una API estándar para el procesamiento de archivos de clase, reduciendo la dependencia de bibliotecas de terceros2.

**Soporte automático**: Las herramientas y frameworks que usen esta API soportarán automáticamente los archivos de clase de las últimas versiones del JDK1.

**Aplicaciones**

Esta API es particularmente útil para:

Frameworks y bibliotecas que necesitan manipular bytecode.
Herramientas de análisis de código.
Generación dinámica de clases en tiempo de ejecución.

Es decir, es probable que esta característica no vaya a ser útil para un developer normal si no para creadores de frameworks, como si hubiera pocos.
Con suerte, veremos integrada esta feature en una RELEASE de spring-boot cuando adopten el JDK23, al igual que la anterior feature.

Estado actual

La Class-File API es una característica en preview en el JDK 23, lo que significa que su diseño y especificación están completos, pero puede cambiar en futuras versiones de Java1.

6. Cuerpos de Constructor Flexibles (JEP 482)
**
**Objetivo

El objetivo principal de esta JEP es permitir que los cuerpos de los constructores contengan declaraciones antes de la invocación explícita de otro constructor (super(...) o this(...)). Esto facilita la inicialización de campos en la misma clase antes de llamar a un constructor de la superclase, lo que puede ser crucial para evitar que el código en el constructor de la superclase acceda a valores predeterminados (como 0, false o null) de los campos en la subclase.

**Cambios Clave**
Estructura del Cuerpo del Constructor: Se permite que el cuerpo del constructor contenga declaraciones antes de una invocación explícita, lo que se denomina prologue. Las declaraciones después de la invocación se denominan epilogue.

    ConstructorBody:
        { [BlockStatements] ExplicitConstructorInvocation [BlockStatements] }

**Inicialización Previo a la Invocación**: Los constructores pueden inicializar campos antes de llamar a super(...), permitiendo así que los métodos en la superclase operen sobre valores ya inicializados.

**Validación y Manejo de Errores**: Permite realizar validaciones y lanzar excepciones antes de llamar al constructor de la superclase, lo que mejora el manejo de errores y permite un estilo más limpio y directo.

**Ejemplo Práctico**

Consideremos un ejemplo donde queremos validar un valor antes de pasarle al constructor de la superclase:

    class Vehicle {
        private final int value;

        Vehicle(int value) {
            this.value = value;
        }
    }

    class EngineValueSensor extends Vehicle {
        private final String type = "engine_sensor";

        EngineValueSensor(Integer value) {
            if (value <= 0) {
                throw new IllegalArgumentException("Value must be greater than zero: " + value);
            }
            super(value); // Llamada al constructor de Vehicle
        }
    }

En este ejemplo, si el valor es inválido, lanzamos una excepción antes de llamar al constructor super(value), asegurando así que no se cree un objeto con un estado inválido.

**Beneficios**

**Simplicidad y Legibilidad**: Reduce la necesidad de métodos auxiliares para preparar argumentos antes de las llamadas al constructor.

**Mejor Mantenimiento**: Facilita el mantenimiento del código al permitir una estructura más lógica y menos fragmentada.

**Flexibilidad**: Permite patrones de diseño más simples y efectivos, como el patrón Factory, donde se pueden crear instancias con configuraciones complejas sin complicaciones adicionales.

*7. Comentarios en Markdown para Documentación (JEP 467)
*

/// Example Class
    /// 
    /// This class demonstrates the use of Markdown in JavaDoc comments.
    /// 
    /// #Features
    /// - Easy to read and write
    /// - Supports Markdown syntax for formatting
    /// 
    /// ##Methods
    /// 
    /// - `void exampleMethod(int param)` - An example method that does something.
    /// 
    /// #Usage
    /// To use this class, create an instance and call the method:
    ///
    /// ```

java
    /// ExampleClass example = new ExampleClass();
    /// example.exampleMethod(10);
    ///

/// 
/// ##Note
/// This feature simplifies documentation by allowing Markdown syntax instead of HTML.
/// 
/// @param param an integer parameter for the example method
public class ExampleClass {
    /// This is an example method.
    /// 
    /// @param param an integer parameter for the method
    public void exampleMethod(int param) {
        // Implementation goes here
    }
}




**  Explicación del Código
**  
    **Encabezados y Formato**:
    Se utilizan encabezados (#, ##, ###) para estructurar la documentación, lo que facilita la lectura.
    Se pueden crear listas usando guiones (-) para enumerar características.

    **Código en Bloques**:
    El uso de tres acentos graves (

) permite incluir ejemplos de código, lo que es útil para mostrar cómo utilizar la clase o método.

** Parámetros y Notas:
** Se pueden incluir descripciones de parámetros utilizando la etiqueta @param, manteniendo la compatibilidad con las convenciones de JavaDoc.

** Markdown vs. HTML:
** Este enfoque evita la complejidad del HTML, haciendo que los comentarios sean más accesibles y fáciles de mantener.

**Ventajas**:
    Simplicidad: La sintaxis Markdown es más fácil de escribir y leer en comparación con HTML.

    Compatibilidad: Las etiquetas JavaDoc tradicionales siguen siendo utilizables dentro de los comentarios en Markdown.

    Flexibilidad: Permite una mejor presentación de la documentación, incluyendo listas, enlaces y bloques de código.

**Conclusión**:

    El uso de comentarios en Markdown para documentación en Java (JEP 467) mejora significativamente la experiencia del desarrollador al escribir y mantener documentación. Este enfoque permite a los desarrolladores aprovechar la simplicidad y claridad del Markdown mientras mantienen la funcionalidad completa de JavaDoc.

Stream Gatherers (JEP 473)

import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import java.util.stream.Gatherers;

public class StreamGatherersExample {
    public static void main(String[] args) {
        // Generar un flujo de números enteros
        List<List<Integer>> sample = Stream.iterate(0, i -> i + 2)
            .gather(Gatherers.windowFixed(2)) // Usar un gatherer para agrupar elementos en ventanas fijas
            .limit(5) // Limitar el flujo a 5 grupos
            .collect(Collectors.toList()); // Recoger el resultado en una lista

        // Imprimir el resultado
        System.out.println(sample);
    }
}

javac --enable-preview --release 23 StreamGatherersExample.java
Note: StreamGatherersExample.java uses preview features of Java SE 23.
Note: Recompile with -Xlint:preview for details.

java --enable-preview StreamGatherersExample
[[0, 2], [4, 6], [8, 10], [12, 14], [16, 18]]

Explicación del Código

Importaciones:
Se importan las clases necesarias para trabajar con flujos y colectores.

Generación del Flujo:
Stream.iterate(0, i -> i + 2) genera un flujo infinito de números enteros comenzando desde 0 y aumentando de 2 en 2.

Uso del Gatherer:
    gather(Gatherers.windowFixed(2)) utiliza un gatherer predefinido que agrupa los elementos en ventanas fijas de tamaño 2. Esto significa que cada grupo contendrá dos elementos del flujo original.

Limitación del Flujo:
    limit(5) restringe el flujo a solo 5 grupos, lo que resulta en un total de 10 elementos.

Colección del Resultado:
collect(Collectors.toList()) convierte el flujo resultante en una lista de listas.

Impresión del Resultado:
Finalmente, se imprime la lista resultante, que debería mostrar grupos de números enteros.

Salida Esperada
Al ejecutar el código anterior, la salida será:
text
[[0, 2], [4, 6], [8, 10], [12, 14], [16, 18]]

Esto indica que los números han sido agrupados correctamente en ventanas de tamaño 2.

Conclusión

El uso de Stream Gatherers permite a los desarrolladores crear operaciones intermedias personalizadas que no son fácilmente alcanzables con las operaciones integradas existentes. Esto proporciona mayor flexibilidad y expresividad al trabajar con flujos de datos en Java.

*Z Garbage Collector (ZGC) en Modo Generacional por Defecto (JEP 474)
*
java -XX:+UseZGC -XX:+ZGenerational -jar tu_aplicacion.trabajar

Netcat pivot relay

@alonso_isidoro — Wed, 20 Jul 2022 10:17:55 +0000

The idea is that you are redirecting traffic from a port not controlled by the firewall (40) to one that is controlled by the firewall (23),ie you want to send something to the port closed by the firewall. To do this, once you have access to the destination machine, you are going to raise a netcat service listening for 23.

> nc -lvp 23
listening on [any] 23 ...
connect to [127.0.0.1] from localhost [127.0.0.1] 42662
hello
redirecting traffic from port 40 that will be outside the control of the firewall to port 23 that will be controlled by the firewall

Then, on another machine or on the same vulnerable machine, you are going to create a file of special characters (pivot) that will serve as a stack where to send the exploit

man mknod
NAME
mknod - make block or character special files
...


> mknod pivot p

Create the bridge from unfiltered port 40 to fw-filtered port 23, using the stack.
When I write to port 40, I read from the stack, when I read from 23, I write to the stack.

> nc -lvp 40 0<pivot | nc 127.0.0.1 23 > pivot
listening on [any] 40 ...
connect to [127.0.0.1] from localhost [127.0.0.1] 44386

Finally create the connection to the vulnerable port.
What you write here is eventually written to the port supposedly filtered by the firewall.

> nc 127.0.0.1 40
hello
redirecting traffic from port 40 that will be outside the control of the firewall to port 23 that will be controlled by the firewall.
This could be some exploit to vuln port 23...

Thanks Santiago ;)