Exporter arguments are always converted to maps instead of list

[Zelldon]

Describe the bug

See for example this configuration:

zeebe:
  broker:
    gateway:
      # Enable the embedded gateway to start on broker startup.
      # This setting can also be overridden using the environment variable ZEEBE_BROKER_GATEWAY_ENABLE.
      enable: true

      network:
        # Sets the port the embedded gateway binds to.
        # This setting can also be overridden using the environment variable ZEEBE_BROKER_GATEWAY_NETWORK_PORT.
        port: 26500

      security:
        # Enables TLS authentication between clients and the gateway
        # This setting can also be overridden using the environment variable ZEEBE_BROKER_GATEWAY_SECURITY_ENABLED.
        enabled: false

    network:
      # Controls the default host the broker should bind to. Can be overwritten on a
      # per binding basis for client, management and replication
      # This setting can also be overridden using the environment variable ZEEBE_BROKER_NETWORK_HOST.
      host: 0.0.0.0

    data:
      # Specify a list of directories in which data is stored.
      # This setting can also be overridden using the environment variable ZEEBE_BROKER_DATA_DIRECTORIES.
      directories: [ data ]
      # The size of data log segment files.
      # This setting can also be overridden using the environment variable ZEEBE_BROKER_DATA_LOGSEGMENTSIZE.
      logSegmentSize: 512MB
      # How often we take snapshots of streams (time unit)
      # This setting can also be overridden using the environment variable ZEEBE_BROKER_DATA_SNAPSHOTPERIOD.
      snapshotPeriod: 15m

    cluster:
      # Specifies the Zeebe cluster size.
      # This can also be overridden using the environment variable ZEEBE_BROKER_CLUSTER_CLUSTERSIZE.
      clusterSize: 1
      # Controls the replication factor, which defines the count of replicas per partition.
      # This can also be overridden using the environment variable ZEEBE_BROKER_CLUSTER_REPLICATIONFACTOR.
      replicationFactor: 1
      # Controls the number of partitions, which should exist in the cluster.
      # This can also be overridden using the environment variable ZEEBE_BROKER_CLUSTER_PARTITIONSCOUNT.
      partitionsCount: 1

    threads:
      # Controls the number of non-blocking CPU threads to be used.
      # WARNING: You should never specify a value that is larger than the number of physical cores
      # available. Good practice is to leave 1-2 cores for ioThreads and the operating
      # system (it has to run somewhere). For example, when running Zeebe on a machine
      # which has 4 cores, a good value would be 2.
      # This setting can also be overridden using the environment variable ZEEBE_BROKER_THREADS_CPUTHREADCOUNT
      cpuThreadCount: 2
      # Controls the number of io threads to be used.
      # This setting can also be overridden using the environment variable ZEEBE_BROKER_THREADS_IOTHREADCOUNT
      ioThreadCount: 2


    exporters:
      kafka:
        className: io.zeebe.exporters.kafka.KafkaExporter
        args:
          maxInFlightRecords: 1000
          inFlightRecordCheckIntervalMs: 1000
          producer:
            servers: ['localhost:9092']
            requestTimeoutMs: 5000
            closeTimeoutMs: 5000
            clientId: zeebe
            maxConcurrentRequests: 3
            config: {}
          records:
            defaults:
              type: [event]
              topic: zeebe
            deployment:
              topic: zeebe-deployment
            incident:
              topic: zeebe-incident
            jobBatch:
              topic: zeebe-job-batch
            job:
              topic: zeebe-job
            message:
              topic: zeebe-message
            messageSubscription:
              topic: zeebe-message-subscription
            messageStartEventSubscription:
              topic: zeebe-message-subscription-start-event
            raft:
              topic: zeebe-raft
            timer:
              topic: zeebe-timer
            variable:
              topic: zeebe-variable
            workflowInstance:
              type: [event]
              topic: zeebe-workflow
            workflowInstanceSubscription:
              topic: zeebe-workflow-subscription

This will be converted and printed later by the broker to:

07:26:48.122 [] [main] INFO  io.zeebe.broker.system - Starting broker 0 with configuration {
  "network" : {
    "host" : "0.0.0.0",
    "portOffset" : 0,
    "maxMessageSize" : "4MB",
    "advertisedHost" : "0.0.0.0",
    "commandApi" : {
      "host" : "0.0.0.0",
      "port" : 26501,
      "advertisedHost" : "0.0.0.0",
      "advertisedPort" : 26501,
      "address" : "0.0.0.0:26501",
      "advertisedAddress" : "0.0.0.0:26501"
    },
    "internalApi" : {
      "host" : "0.0.0.0",
      "port" : 26502,
      "advertisedHost" : "0.0.0.0",
      "advertisedPort" : 26502,
      "address" : "0.0.0.0:26502",
      "advertisedAddress" : "0.0.0.0:26502"
    },
    "monitoringApi" : {
      "host" : "0.0.0.0",
      "port" : 9600,
      "advertisedHost" : "0.0.0.0",
      "advertisedPort" : 9600,
      "address" : "0.0.0.0:9600",
      "advertisedAddress" : "0.0.0.0:9600"
    },
    "maxMessageSizeInBytes" : 4194304
  },
  "cluster" : {
    "initialContactPoints" : [ ],
    "partitionIds" : [ 1 ],
    "nodeId" : 0,
    "partitionsCount" : 1,
    "replicationFactor" : 1,
    "clusterSize" : 1,
    "clusterName" : "zeebe-cluster",
    "gossipFailureTimeout" : 10000,
    "gossipInterval" : 250,
    "gossipProbeInterval" : 1000
  },
  "threads" : {
    "cpuThreadCount" : 2,
    "ioThreadCount" : 2
  },
  "data" : {
    "directories" : [ "/home/zell/goPath/src/github.com/zeebe-io/zeebe/data" ],
    "logSegmentSize" : "512MB",
    "snapshotPeriod" : "PT15M",
    "logIndexDensity" : 100,
    "logSegmentSizeInBytes" : 536870912,
    "atomixStorageLevel" : "DISK"
  },
  "exporters" : {
    "kafka" : {
      "jarPath" : null,
      "className" : "io.zeebe.exporters.kafka.KafkaExporter",
      "args" : {
        "maxInFlightRecords" : 1000,
        "inFlightRecordCheckIntervalMs" : 1000,
        "producer" : {
          "servers" : {
            "0" : "localhost:9092"
          },
          "requestTimeoutMs" : 5000,
          "closeTimeoutMs" : 5000,
          "clientId" : "zeebe",
          "maxConcurrentRequests" : 3
        },
        "records" : {
          "defaults" : {
            "type" : {
              "0" : "event"
            },
            "topic" : "zeebe"
          },
          "deployment" : {
            "topic" : "zeebe-deployment"
          },
          "incident" : {
            "topic" : "zeebe-incident"
          },
          "jobBatch" : {
            "topic" : "zeebe-job-batch"
          },
          "job" : {
            "topic" : "zeebe-job"
          },
          "message" : {
            "topic" : "zeebe-message"
          },
          "messageSubscription" : {
            "topic" : "zeebe-message-subscription"
          },
          "messageStartEventSubscription" : {
            "topic" : "zeebe-message-subscription-start-event"
          },
          "raft" : {
            "topic" : "zeebe-raft"
          },
          "timer" : {
            "topic" : "zeebe-timer"
          },
          "variable" : {
            "topic" : "zeebe-variable"
          },
          "workflowInstance" : {
            "type" : {
              "0" : "event"
            },
            "topic" : "zeebe-workflow"
          },
          "workflowInstanceSubscription" : {
            "topic" : "zeebe-workflow-subscription"
          }
        }
      },
      "external" : false
    }
  },
  "gateway" : {
    "network" : {
      "host" : "0.0.0.0",
      "port" : 26500,
      "minKeepAliveInterval" : "PT30S"
    },
    "cluster" : {
      "contactPoint" : "0.0.0.0:26502",
      "requestTimeout" : "PT15S",
      "clusterName" : "zeebe-cluster",
      "memberId" : "gateway",
      "host" : "0.0.0.0",
      "port" : 26502
    },
    "threads" : {
      "managementThreads" : 1
    },
    "monitoring" : {
      "enabled" : false,
      "host" : "0.0.0.0",
      "port" : 9600
    },
    "security" : {
      "enabled" : false,
      "certificateChainPath" : null,
      "privateKeyPath" : null
    },
    "enable" : true
  },
  "backpressure" : {
    "enabled" : true,
    "algorithm" : "VEGAS"
  },
  "stepTimeout" : "PT5M",
  "executionMetricsExporterEnabled" : false
}

The problematic parts are

          producer:
            servers: ['localhost:9092']

or

          producer:
            servers: 
               - 'localhost:9092'

Which are converted to:

        "producer" : {
          "servers" : {
            "0" : "localhost:9092"
          },

Which is unexpected.

To Reproduce

Just use the configuration above.

Expected behavior
A list instead of a map.

Reported via https://stackoverflow.com/questions/61826566/converting-yaml-to-java-object

[npepinpe]

I did a prototype where you can grab the Spring context to later instantiate beans programmatically. It's a little clunky since it's not a normal use case (in that respect Quarkus Config/Small Rye is a bit nicer because that's an expected use case), but it's do-able and would be a better replacement. I would propose to consider it as we refactor the bootstrapping - not that we resolve this, but we think about making the ApplicationContext available and then we can easily implement a solution here afterwards.

[npepinpe]

I think this issue #7628 would solve this problem.

[menski]

@menski to check if this is resolved by the issue linked above

[npepinpe]

It's not, since Spring Boot is the layer which partially deserializes the args to to a Map<String, Object> (with the nested lists becoming maps with indexes). We didn't use the application context to parse things in the end, so it wasn't solved.

[saig0]

Marking priority as later later because custom exporters are not our priority at the moment. We focus more on SaaS.

Please comment if you think this should have a higher priority.