Skip to content
View as markdown Ask ChatGPT Ask Claude

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

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.

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 <systemname>_iotevent_dataplatform_*
Factory Automation <systemname>_iotevent_factoryautomation_*
Enterprise Integration <systemname>_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#

The driver forwards a message that it cannot route to the <systemname>_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 API Post Event.

Topic name change in version 12.0.0

Before version 12.0.0, the driver forwarded these messages to the <systemname>_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.

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://<address>:<port>).
  • TLS Connection - use the amqps:// scheme in rabbitMQAddress (amqps://<address>:<port>).

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.