What does Atlan crawl from Confluent Kafka?
During a Confluent Kafka crawl, Atlan extracts metadata for Clusters, Topics, Consumer Groups, Schema Subjects, Schema Versions, and Schema Fields, and can build lineage from Kafka Connect connectors. Reference tables map each Confluent Kafka property to its Atlan asset.
Atlan extracts metadata for Clusters, Topics, Consumer Groups, Schema Subjects, Schema Versions, and Schema Fields from Confluent Kafka and maps each property to its internal asset model. When you turn on Connect lineage, Atlan also creates lineage processes from your Kafka Connect connectors. The tables below list which Confluent Kafka properties are read and how they appear on the corresponding Atlan asset. Once you've crawled Confluent Kafka, you can use connector-specific filters for quick asset discovery.
Cluster
Atlan maps the Kafka cluster to its KafkaCluster asset type. One cluster asset is created per connection. You can filter clusters by bootstrap servers and authentication type.
| Source property | Atlan property |
|---|---|
| Cluster name | name |
| Bootstrap server addresses | kafkaBootstrapServers |
| Total topic count | kafkaTopicCount |
| Total partition count | kafkaPartitionCount |
| Total broker count | kafkaBrokerCount |
| Authentication mechanism | kafkaAuthenticationType |
| Schema Registry URL (if configured) | kafkaSchemaRegistryUrl |
Topics
Atlan maps topics from Confluent Kafka to its KafkaTopic asset type. You can filter topics by message count, size (MB), partition count, and cleanup policy.
| Source property | Atlan property |
|---|---|
| Topic | name |
| PartitionCount | kafkaTopicPartitionsCount |
| ReplicationFactor | kafkaTopicReplicationFactor |
| segment.bytes | kafkaTopicSegmentBytes |
| compression.type | kafkaTopicCompressionType |
| cleanup.policy | kafkaTopicLogCleanupPolicy |
| isInternal | kafkaTopicIsInternal |
| sizeInBytes | kafkaTopicSizeInBytes |
| recordCount | kafkaTopicRecordCount |
| retention.ms | kafkaTopicRetentionTimeInMs |
| retention.bytes | kafkaTopicRetentionBytes |
| Schema Registry subject name (if Schema Registry configured) | kafkaTopicSchemaRegistrySubjectName |
| Whether topic is governed by a schema registry | kafkaTopicIsSchemaManaged |
| Number of active consumer groups | kafkaTopicConsumerCount |
| Parent cluster qualified name | kafkaTopicClusterQualifiedName |
Retrieving sizeInBytes requires a Cloud API key with the MetricsViewer role, Enable Cloud API set to True in the credential, and Include Cloud Metrics set to True in the workflow. Otherwise, sizeInBytes is set to 0.
Consumer groups
Atlan maps consumer groups from Confluent Kafka to its KafkaConsumerGroup asset type. You can filter consumer groups by member count and topic name. Consumer groups appear only when they're active at the time of crawling. If a topic isn't being consumed actively, Confluent Kafka purges the consumer group and it won't be cataloged as an asset.
| Source property | Atlan property |
|---|---|
| GROUP | name |
| memberCount | kafkaConsumerGroupMemberCount |
| topic_names | kafkaTopicNames |
| TOPIC | kafkaConsumerGroupTopicConsumptionProperties.topicName |
| PARTITION | kafkaConsumerGroupTopicConsumptionProperties.topicPartition |
| LAG | kafkaConsumerGroupTopicConsumptionProperties.topicLag |
| CURRENT-OFFSET | kafkaConsumerGroupTopicConsumptionProperties.topicCurrentOffset |
| Consumer group state | kafkaConsumerGroupState |
| Assigned topic-partition pairs | kafkaConsumerGroupAssignedPartitions |
The consumerGroupState property reflects the current state of the consumer group at the time of crawling. Possible values are:
| Value | Description |
|---|---|
Stable | The group has active members with all partitions assigned. |
Empty | The group exists but has no active members. |
PreparingRebalance | The group is collecting member join requests ahead of a rebalance. |
CompletingRebalance | The group is waiting for members to receive their partition assignments. |
Dead | The group has no members and its metadata has been removed. |
Schema subjects
Atlan maps subjects from a connected Schema Registry to its SchemaRegistrySubject asset type. Schema subjects are crawled only when Schema Registry credentials are provided. You can filter schema subjects by schema type and compatibility.
| Source property | Atlan property |
|---|---|
| subject | name |
| schemaType | schemaRegistrySchemaType |
| id | schemaRegistrySchemaId |
| subject (base name without -key/-value suffix) | schemaRegistrySubjectBaseName |
subject ends with -key | schemaRegistrySubjectIsKeySchema |
| compatibilityLevel | schemaRegistrySubjectSchemaCompatibility |
| version (latest) | schemaRegistrySubjectLatestSchemaVersion |
| schema (latest definition) | schemaRegistrySubjectLatestSchemaDefinition |
Schema versions
Atlan maps each schema version to its SchemaRegistryVersion asset type, allowing you to track the full version history of a subject. Schema versions are crawled only when Schema Registry credentials are provided. You can filter schema versions by version number and schema type.
| Source property | Atlan property |
|---|---|
| version number | schemaRegistryVersionNumber |
| schema definition | schemaRegistryVersionSchemaDefinition |
| schema type (AVRO, JSON, PROTOBUF) | schemaRegistryVersionSchemaType |
| parent subject qualified name | schemaRegistrySubjectQualifiedName |
Schema fields
Atlan maps fields from schema definitions to its KafkaField asset type. Atlan doesn't create field-level lineage for these fields. Schema fields are crawled only when Schema Registry credentials are provided. You can filter schema fields by data type and whether they're optional or required.
| Source property | Atlan property |
|---|---|
| field name | name |
| field data type | kafkaFieldDataType |
| field is optional | kafkaFieldIsOptional |
| default value | kafkaFieldDefaultValue |
| field position in schema (0-based) | kafkaFieldOrder |
| parent topic qualified name | kafkaTopicQualifiedName |
| schema type (key or value schema) | kafkaFieldSchemaType |
Connect lineage
When you turn on Include Connect Lineage, Atlan reads your Kafka Connect connector configurations through the Confluent Cloud API and maps each Snowflake sink connector to Process assets. Each process links the Kafka topics that a connector reads to the Snowflake table it writes to. Atlan keeps only an allowlist of non-secret configuration keys needed to trace topics to tables, such as the connector class, topic, transform, and target database, schema, and table keys. It never stores connector credentials.
| Source property | Atlan property |
|---|---|
| Connector name, connector class, and target table | name, as <connector name> (<connector class>) -> <DATABASE>.<SCHEMA>.<TABLE> |
| Topics the connector reads | inputs |
| Snowflake target table | outputs |
Atlan creates one process for each connector and target table. If several topics feed the same table, that process has several inputs.
Atlan resolves the target table from the connector configuration:
- The database comes from
snowflake.database.nameand the schema fromsnowflake.schema.name. - The table name comes from
snowflake.topic2table.mapwhen the topic has an entry. Map keys match the topic exactly first, then as a regular expression. If a map value includes a schema or database, for exampleSCHEMA.TABLE, it overrides the connector-level setting. - When no map entry matches, Atlan derives the table name from the topic name the same way the Snowflake connector does, in uppercase.
The following limits apply:
- Only Snowflake sink connectors produce lineage. Atlan reads other connectors, such as JDBC, CDC, Salesforce, and S3 connectors, but doesn't create processes for them.
- Atlan links lineage only to Snowflake tables that already exist in Atlan. If a target table hasn't been crawled, Atlan skips that process.
- If the same table exists in more than one Snowflake connection, Atlan links the lineage to only one of those connections.
- Lineage is at the table level. Connector configurations don't carry column mappings, so Atlan doesn't create column-level lineage.
- Turning off Include Connect Lineage removes the Connect lineage on the next run.