--- alias: user-guide-driver-factoryautomation tags: - connect iot description: "This documentation details configuration parameters and data types for factory automation driver connectivity" --- # Connect IoT - Factory Automation Driver The Factory Automation driver integrates ConnectIoT with the MES Factory Automation runtime. It acts as a bridge between Kafka-based IoT events and the automation job lifecycle managed by the MES and RabbitMQ. The driver subscribes to system IoT event topics, resolves the corresponding automation controller configuration, creates or synchronizes automation jobs, and dispatches those jobs through the message queue for execution. This implementation relies on: - Kafka as the event ingestion layer - RabbitMQ as the job queue for controller execution - MES System API and Message Bus for controller resolution and job state synchronization ## Data Types These are the supported specific data types: | Name | Description | | ------- | ---------------- | | String | String Value | | Boolean | Boolean Value | | Integer | Integer value | | Decimal | Decimal value | | Object | Structured value | Table: Factory Automation specific data types ## Protocol Parameters The protocol supports the following parameters: | Name | Type | Possible Values | Default Value | Description | | ------------------------------ | -------- | --------------------------------------------------------- | ----------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | bootstrapServers | String | | kafka1:9092,kafka2:9092,kafka3:9092 | This is a comma-separated list of host and port pairs that are the addresses of the Kafka brokers in a "bootstrap" Kafka cluster that a Kafka client connects to initially to bootstrap itself | | kafkaAuthenticationMethod | Enum | None, SASL_Plain, SASL_SSL Plain, SASL_SSL SCRAM512, mTLS | None | Security protocol used | | kafkaUserName | String | | | User name to use for authentication in Kafka for SASL authentication | | kafkaPassword | Password | | | Password to use for authentication in Kafka for SASL authentication | | kafkaCaPem | Text | | | SSL CA Certificate Pem | | kafkaCertificatePem | Text | | | SSL Pem Certificate | | kafkaKeyPem | Text | | | SSL Private Key | | kafkaValidateCertificates | Boolean | | True | Validates the certificates | | topicMetadataRefreshIntervalMs | Integer | 1 - 3600000 | 300000 | Interval, in milliseconds, for refreshing Kafka topic metadata. It controls how quickly a newly created IoT event topic is discovered by the driver | | fromBeginning | Boolean | | False | Use the earliest retained offset when a partition has no committed consumer-group offset. Existing committed offsets always take precedence | | rabbitMQAddress | String | | amqp://rabbitmq:5672 | RabbitMQ address, including the scheme that selects the connection security (`amqp` for a plaintext connection or `amqps` for a TLS connection) | | rabbitMQUseClientCertificates | Boolean | | False | Currently not used. The connection security is derived from the `rabbitMQAddress` scheme and the certificate parameters instead | | rabbitMQSecurityProtocol | Enum | Plaintext, Ssl | Plaintext | Currently not used. The connection security is determined by the `rabbitMQAddress` scheme (`amqp`/`amqps`) instead | | rabbitMQUserName | String | | | User name to use for authentication in RabbitMQ | | rabbitMQPassword | Password | | | Password to use for authentication in RabbitMQ | | rabbitMQCertificatePem | Text | | | SSL Pem Certificate | | rabbitMQKeyPem | Text | | | SSL Private Key | | rabbitMQCaPem | Text | | | SSL CA Certificate Pem | | rabbitMQValidateCertificates | Boolean | | True | Validates the certificates | | maxConcurrentJobs | Integer | | 10 | Maximum number of jobs that can be running at the same time | | controllerCommunicationTimeout | Integer | | 5000 | Maximum amount of time to allow the controller to reply to requests from the driver (ms) | | retryAttempts | Integer | | 30 | Maximum number of retries to perform any job related communication | | sleepBetweenAttempts | Integer | | 1000 | Time to wait between retries to perform any job related communication | | systemPollingInterval | Integer | | 300000 | Time between calls to the system for updates on child jobs | | queueName | String | | ${SystemName}_IoT_Job | The name of the queue to use. The only available token to be used is SystemName (This token is replaced by the System Name, example: MES_IoT_Job) | Table: Factory Automation specific parameters !!! warning "Behavior change in version 12.0.0" Before version 12.0.0, a consumer group without a committed offset always started at the earliest retained offset. Starting with version 12.0.0, the `fromBeginning` parameter controls this behavior and defaults to False, so a consumer group without a committed offset starts at the latest offset and skips the messages retained before it joins. This also applies to the topic of a new IoT Event Definition, so the messages posted to it before the driver discovers it are skipped. To keep the previous behavior, enable `fromBeginning` in the Automation Protocol. !!! note "Job queue type" The driver always declares its job queue (`queueName`) as a **classic** RabbitMQ queue with a maximum priority of 10, regardless of the virtual host's `default_queue_type` setting. If the RabbitMQ server is configured to default new queues to **quorum** type, the driver's explicit `classic` declaration still takes precedence for this queue. This is not currently configurable. ## Event Topics The driver consumes IoT events from Kafka through one subscription per IoT Event Definition scope. Each subscription is a regular expression built from the system name, so every topic of that scope is matched, including topics created after the driver connects: | Scope | Subscribed topics | | ---------------------- | ----------------------------------------------- | | Data Platform | `_iotevent_dataplatform_*` | | Factory Automation | `_iotevent_factoryautomation_*` | | Enterprise Integration | `_iotevent_enterpriseintegration_*` | Table: Factory Automation driver event topic subscriptions The system name is lowercased before the topic names are built. The three subscriptions share a single consumer group. A new IoT Event Definition creates a new topic, which the driver picks up on the next topic metadata refresh. Lower `topicMetadataRefreshIntervalMs` for shorter discovery delay, at the cost of additional metadata traffic. Consumption is paused until the driver completes its setup. Messages that arrive before that are not committed, so they are consumed again once the driver is ready. A message that cannot be routed is forwarded to the dead-letter queue. For details, see [Dead-Letter Queue](#dead-letter-queue). ## Dead-Letter Queue The driver forwards a message that it cannot route to the `_dp_iotevents_dlq` topic. There is a single dead-letter queue, shared by the three scopes, and the system name is lowercased before the topic name is built. The driver forwards a message in the following cases: - The message has no `IoTEventDefinitionName` header, or the header is empty. - The message has an `IoTEventDefinitionId` header that does not match the ID of the IoT Event Definition known to the driver for that name. - The IoT Event Definition is not found in the MES. The message is forwarded with its original key, headers, and value. The driver logs an error each time it forwards a message to the dead-letter queue, and then commits the offset of the message, so the message is not consumed again. The MES publishes IoT events that fail to post to the same topic. For details, see [[user-guide-automation-task-core-apipostevent]]. !!! note "Topic name change in version 12.0.0" Before version 12.0.0, the driver forwarded these messages to the `_dp_dataplatform_raw_dlq` topic. ## Connection Security ### Kafka The Kafka connection security is determined by the Kafka Authentication Method (`kafkaAuthenticationMethod`) parameter: - **None** - no authentication and no encryption. - **SASL_Plain** - SASL `PLAIN` authentication over a plaintext connection, using `kafkaUserName` and `kafkaPassword`. - **SASL_SSL Plain** - SASL `PLAIN` authentication over a Transport Layer Security (TLS) connection, using `kafkaUserName` and `kafkaPassword`. - **SASL_SSL SCRAM512** - SASL `SCRAM-SHA-512` authentication over a TLS connection, using `kafkaUserName` and `kafkaPassword`. - **mTLS** - mutual TLS authentication, using the client certificate provided through `kafkaCertificatePem` and `kafkaKeyPem`. When a secure (TLS) method is used (either `SASL_SSL` method or `mTLS`), specify the following certificate-related parameters: - `kafkaCaPem` - confirms the server is genuine. Only required when the broker's certificate is not issued by a publicly trusted certificate authority (for example, a self-signed or private CA); otherwise the system's trusted certificate store is used. - `kafkaCertificatePem` and `kafkaKeyPem` - prove who you are to the server. These are only needed for `mTLS`. - `kafkaValidateCertificates` - turns the server identity check on or off. It only has an effect when a secure method is selected. ### RabbitMQ The RabbitMQ connection security is determined by the scheme set in `rabbitMQAddress` and the presence of certificate parameters. !!! note "Deprecated Parameters" The `rabbitMQSecurityProtocol` and `rabbitMQUseClientCertificates` parameters are not used by the driver. Changing them will not alter connection behavior. To configure security, use the appropriate address format: - **Plaintext Connection** - use the `amqp://` scheme in `rabbitMQAddress` (`amqp://
:`). - **TLS Connection** - use the `amqps://` scheme in `rabbitMQAddress` (`amqps://
:`). When a secure (TLS) connection (`amqps`) is active, manage certificates using these parameters: - `rabbitMQCaPem` - confirms the server is genuine. Only required when the broker's certificate is not issued by a publicly trusted certificate authority (for example, a self-signed or private CA); otherwise the system's trusted certificate store is used. - `rabbitMQCertificatePem` and `rabbitMQKeyPem` - prove who you are to the server. These are only needed when the broker requires certificate-based client authentication. - `rabbitMQValidateCertificates` - turns the server identity check on or off. It only has an effect when the `amqps` scheme is used.