K6: Evaluando la Performance de los Consumers en Kafka: Estrés en los Tópicos 🚀

import { check, fail } from "k6";
import { Counter } from 'k6/metrics';
import { Writer, SchemaRegistry, SCHEMA_TYPE_STRING } from "k6/x/kafka";

const success = new Counter('kafka_success');
const timeout = new Counter('kafka_timeout_errors');
const unexpectedErrors = new Counter('kafka_Unexpected_errors');
const connectionRefusedErrors = new Counter('kafka_connection_refused_errors');
const ioPipeErrors = new Counter('kafka_io_pipe_errors');
const response_time_2s = new Counter('kafka_response_time_2s');
const topic_or_partition  = new Counter('kafka_topic_or_partition_not_exist');

const schemaRegistry = new SchemaRegistry();

export let options = {
    insecureSkipTLSVerify: true, 
    scenarios: {
        ramp_up_and_constant_rate: {
            executor: 'ramping-arrival-rate',
            startRate: 1,
            timeUnit: '1s',
            preAllocatedVUs: 3,
            maxVUs: 10,
            stages: [
                { duration: '2m', target: 1 },
            ],
        },
    },
};

export default function () {

const writer = new Writer({
    brokers: ["localhost:9092"],
    topic: "miTopico",
});

    const message = { syskey: "test-key", foo: "bar" };

    const key = schemaRegistry.serialize({ data: message.syskey, schemaType: SCHEMA_TYPE_STRING });
    const value = schemaRegistry.serialize({ data: JSON.stringify(message), schemaType: SCHEMA_TYPE_STRING });

    try {
        
        const start = Date.now(); //inicia cronometro
        
        const error = writer.produce({
            messages: [{ key, value }],
        });
        
        // writer.produce() retorna "undefined" cuando todo salio bien! caso contrario, devuelve mensaje de error

        // Opcional:
        //const success = check(error, { "mensaje enviado sin error": (err) => err === undefined,});        

        if (error !== undefined && error.message) {
        
          clasificarError(error.message);
            
        } else {
            success.add(1);
            const end = Date.now(); //termina cronometro
            const produceDuration = (end - start);
            if (produceDuration >= 1500){
              response_time_2s.add(1)
            }          
        }
               
    //K6 lanza una excepcion (GoError)
    } catch (e) {
    
        clasificarError(e.message);
        
        fail(`Fallo la ejecucion: ${e.message}`);
        
    } finally {
        writer.close();
    }
}

function clasificarError(msg) {
    msg = msg.toLowerCase();

    switch (true) {
        case msg.includes("read/write on closed pipe"):
            ioPipeErrors.add(1);
            break;
        case msg.includes("connection refused"):
            connectionRefusedErrors.add(1);
            break;
        case msg.includes("timeout"):
            timeout.add(1);
            break;
        case msg.includes("a topic or partition that does not exist on this broker"):
            topic_or_partition.add(1);
            break;
        default:
            unexpectedErrors.add(1);
            break;
    }
}

import { check } from "k6";
import {TLS_1_2, SASL_SCRAM_SHA512 ,Writer,SchemaRegistry ,SCHEMA_TYPE_STRING} from "k6/x/kafka";
...
const writer = new Writer({
    brokers: ["localhost:9092"],
    topic: "miTopico",
    sasl: {
        username: "NombreUsuario",
        password: "PasswordDelUsuario",
        algorithm: SASL_SCRAM_SHA512,
    },
    tls: {
        clientCertPem: '/home/user/script/kafka_cert.pem', // Path del certificado
        enableTls: true,
        insecureSkipTlsVerify: true,
        minVersion: TLS_1_2
        },
});
...