---
apiVersion: v1
kind: ConfigMap
metadata:
  name: kafka-producer-1
  namespace: k6-tests
data:
  test.js: |
    import { check, sleep } from 'k6';
    import { Writer, Connection, SchemaRegistry, SCHEMA_TYPE_STRING } from 'k6/x/kafka';
    import { Rate } from 'k6/metrics';

    const bootstrapServers = [`${__ENV.KAFKA_BOOTSTRAP}`];
    const kafkaTopic = `${__ENV.TOPIC}`;

    console.log('bootstrapServers:', bootstrapServers, ', kafkaTopic', kafkaTopic);

    const connection = new Connection({
        address: bootstrapServers[0],
    });

    // const topics = connection.listTopics();
    // console.log("Topics: ", topics);

    const schemaRegistry = new SchemaRegistry();

    const writer = new Writer({
      brokers: bootstrapServers, 
      topic: kafkaTopic,
    });

    export const options = {
      scenarios: {
        start_1: {
          executor: 'constant-arrival-rate',
          rate: 1, // 1 RPS, since timeUnit is the default 1s
          timeUnit: '1s',
          duration: '2m',
          preAllocatedVUs: 1,
        },
        ramp_1: {
          executor: 'ramping-arrival-rate',
          startTime: '2m',
          startRate: 1,
          timeUnit: '1s',
          preAllocatedVUs: 1,
          maxVUs: 6,
          stages: [
            { target: 6, duration: '1m' },
            { target: 6, duration: '5m' },
            { target: 1, duration: '1m' },
          ],
        },
        end_1: {
          executor: 'constant-arrival-rate',
          startTime: '8m',
          rate: 1, 
          timeUnit: '1s',
          duration: '2m',
          preAllocatedVUs: 1,
        },
      },
    };

    export default function () {

      const d = new Date();
      const error = writer.produce({
        messages: [
          {
            key: schemaRegistry.serialize({
              data: "test-key-string-"+d.toISOString(),
              schemaType: SCHEMA_TYPE_STRING,
            }),
            value: schemaRegistry.serialize({
              data: "test-value-string-"+d.toISOString(),
              schemaType: SCHEMA_TYPE_STRING,
            }),
            headers: {
              mykey: "myvalue",
            },
          },
        ],
      });

      check(error, {
        'is sent': (err) => err == undefined,
      });
    }

    export function teardown(data) {
      writer.close();
    }
