Supporting catalog targets¶

A catalog target describes a specific piece of third party technology that an integration connector is exchanging metadata with.

Why catalog targets change the way connectors are built¶

Originally, the technology that an integration connector worked with was identified by the endpoint in the connector's own connection. A connection describes one endpoint, so a connector instance could work with exactly one database, one file directory, or one Kafka broker. If you had three databases to catalog, you configured and deployed three instances of the connector, each with its own connection, its own credentials and its own entry in the integration daemon's configuration.

With catalog targets, the connector's connection only needs to identify the connector implementation. The technology to work with is attached to the connector's IntegrationConnector metadata element using a CatalogTarget relationship that points at the asset describing the technology.

The integration connector is linked to the asset for the database it is to catalog. The asset's own connection - including any embedded secrets store connector - is used to create the connector that calls the database.

This has a big effect on operations:

• The connector is deployed once - typically by loading a content pack into the metadata access store at start up - and it then runs in an integration group.
• When someone wants a new database catalogued, they attach the database's asset to the connector with a CatalogTarget relationship. No configuration change, no restart.
• Connection details, including credentials, are defined once on the asset and reused by every connector, survey action service and governance action service that works with that resource.
• Each catalog target can carry its own configuration properties, templates, metadata source name and permitted synchronization, so one running connector can treat each of its targets differently.

It also changes the way the connector is built. The per-resource logic that used to live in the connector's refresh() method moves into a catalog target processor.

The shape of a catalog target connector¶

flowchart TD
    subgraph one["Integration daemon thread"]
    A["`**MyIntegrationConnector**
    _extends DynamicIntegrationConnectorBase_
    start(), refresh(), disconnect()`"]
    end
    A -->|"getNewRequestedCatalogTargetSkeleton()"| B["`**MyCatalogTargetProcessor**
    _extends CatalogTargetProcessorBase_
    catalog target: *sales database*`"]
    A --> C["`**MyCatalogTargetProcessor**
    catalog target: *finance database*`"]
    A --> D["`**MyCatalogTargetProcessor**
    catalog target: *HR database*`"]

There are two classes to write (plus the connector provider):

The base class does the rest. On each refresh() it:

1. Retrieves the current list of CatalogTarget relationships attached to the connector's element.
2. For each new one, creates a digital resource connector for the target asset (if the target is an asset with a connection), calls your getNewRequestedCatalogTargetSkeleton() method to create the processor, overlays the configuration properties, starts the resource connector and calls start() on the processor.
3. For each one whose relationship properties have changed, disconnects the old resource connector and repeats step 2 with the new values.
4. Calls refresh() on every processor, catching and logging exceptions so that one failing target does not stop the others.
5. Registers an open metadata event listener (once the first refresh is complete) if the connector implements OpenMetadataEventListener, and passes each event to any processor that implements OpenMetadataEventListener.

Writing the integration connector¶

The connector extends DynamicIntegrationConnectorBase and implements a single abstract method: getNewRequestedCatalogTargetSkeleton(). Here is the complete implementation of the Kafka Topic Integration Connector :

public class KafkaTopicIntegrationConnector extends DynamicIntegrationConnectorBase
{
    /**
     * Create a new catalog target processor (typically inherits from CatalogTargetProcessorBase).
     *
     * @param retrievedCatalogTarget details of the open metadata elements describing the catalog target
     * @param catalogTargetContext   specialized context for this catalog target
     * @param connectorToTarget      connector to access the target resource
     * @return new processor based on the catalog target information
     */
    @Override
    public RequestedCatalogTarget getNewRequestedCatalogTargetSkeleton(CatalogTarget        retrievedCatalogTarget,
                                                                       CatalogTargetContext catalogTargetContext,
                                                                       Connector            connectorToTarget)
    {
        return new KafkaTopicCatalogTargetProcessor(retrievedCatalogTarget,
                                                    catalogTargetContext,
                                                    connectorToTarget,
                                                    connectorName,
                                                    auditLog);
    }
}

Notice that there is no start(), refresh() or disconnect() method. They are only needed if the connector has work of its own to do. For example, the PostgreSQL Server Integration Connector overrides start() to read the default database include/exclude lists from its own connection:

    @Override
    public void start() throws ConnectorCheckedException, UserNotAuthorizedException
    {
        super.start();

        defaultExcludeDatabases = super.getArrayConfigurationProperty(PostgresConfigurationProperty.EXCLUDE_DATABASE_LIST.getName(),
                                                                      connectionBean.getConfigurationProperties(),
                                                                      Collections.singletonList("postgres"));
        :
    }

Writing the catalog target processor¶

The processor extends CatalogTargetProcessorBase , which in turn extends RequestedCatalogTarget and CatalogTarget. This means the properties of the catalog target are available directly on this:

The start method¶

start() is called once, when the catalog target is first seen (and again if the relationship is updated). This is where to validate the target and extract the values that do not change between refreshes. From the Kafka Topic Catalog Target Processor :

    @Override
    public void start() throws UserNotAuthorizedException, ConnectorCheckedException
    {
        super.start();

        final String methodName = "start";

        integrationContext.validateIsActive(methodName);

        if ((super.connectorToTarget != null) &&
            (super.connectorToTarget.getConnection() != null) &&
            (super.connectorToTarget.getConnection().getEndpoint() != null) &&
            (super.connectorToTarget.getConnection().getEndpoint().getNetworkAddress() != null))
        {
            hostIdentifier = super.getHostIdentifier(super.connectorToTarget.getConnection().getEndpoint().getNetworkAddress());
            portNumber     = super.getPortNumber(super.connectorToTarget.getConnection().getEndpoint().getNetworkAddress());
            templateGUID   = this.getTemplateGUID();
        }
        else
        {
            super.throwMissingConnectionInfo(OpenMetadataProperty.NETWORK_ADDRESS.name, methodName);
        }

        if (propertyHelper.isTypeOf(this.getCatalogTargetElement().getElementHeader(), OpenMetadataType.SOFTWARE_SERVER.typeName))
        {
            eventBrokerGUID = this.getEventBrokerGUID(this.getCatalogTargetElement());
        }
        else if (propertyHelper.isTypeOf(this.getCatalogTargetElement().getElementHeader(), OpenMetadataType.EVENT_BROKER.typeName))
        {
            eventBrokerGUID = this.getCatalogTargetElement().getElementHeader().getGUID();
        }
        else
        {
            super.throwWrongTypeOfCatalogTarget(Arrays.asList(OpenMetadataType.SOFTWARE_SERVER.typeName,
                                                              OpenMetadataType.EVENT_BROKER.typeName).toString(),
                                                methodName);
        }
    }

CatalogTargetProcessorBase supplies a family of helper methods for this validation work, so that all connectors report the same problems in the same way:

• throwWrongTypeOfCatalogTarget() - the element on the end of the relationship is not a type this connector can process.
• throwWrongTypeOfResourceConnector() - the asset's connection created a connector of an unexpected class.
• throwMissingConnectionInfo() - the connection is missing a value the connector needs, such as the network address.
• throwMissingPropertyValue() and throwBadBeanClass() - the target element does not have the properties the connector expects.

It also supplies methods to extract typed values from the combined configuration properties - getStringConfigurationProperty(), getBooleanConfigurationProperty(), getIntConfigurationProperty(), getLongConfigurationProperty(), getDateConfigurationProperty(), getArrayConfigurationProperty() and getSuppliedPlaceholderProperties() - along with getHostIdentifier() and getPortNumber() for splitting a network address.

The refresh method¶

refresh() does the metadata exchange for this one target. Because the base class calls it inside a try/catch, an exception thrown here takes only this target out of action for this pass, not the whole connector.

    @Override
    public void refresh() throws ConnectorCheckedException
    {
        final String methodName = "refresh";

        if (propertyHelper.isTypeOf(this.getCatalogTargetElement().getElementHeader(),
                                    PostgresDeployedImplementationType.POSTGRESQL_SERVER.getAssociatedTypeName()))
        {
            JDBCResourceConnector assetConnector = null;

            try
            {
                String                  databaseServerGUID = this.getCatalogTargetElement().getElementHeader().getGUID();
                OpenMetadataRootElement databaseManager    = this.getDatabaseManager(databaseServerGUID, this.getCatalogTargetElement());

                Connector connector = integrationContext.getConnectedAssetContext().getConnectorForAsset(databaseServerGUID, auditLog);

                assetConnector = (JDBCResourceConnector)connector;
                assetConnector.start();

                catalogDatabases(databaseServerGUID,
                                 databaseManager,
                                 this.getTemplates(),
                                 this.getConfigurationProperties(),
                                 assetConnector);
            }
            catch (Exception exception)
            {
                auditLog.logException(methodName,
                                      PostgresAuditCode.UNEXPECTED_EXCEPTION.getMessageDefinition(connectorName,
                                                                                                  exception.getClass().getName(),
                                                                                                  methodName,
                                                                                                  exception.getMessage()),
                                      exception);
            }
            finally
            {
                this.releaseAssetConnector(assetConnector, methodName);
            }
        }
        else
        {
            super.throwWrongTypeOfCatalogTarget(PostgresDeployedImplementationType.POSTGRESQL_SERVER.getAssociatedTypeName(),
                                                methodName);
        }
    }

The disconnect method¶

disconnect() is called when the CatalogTarget relationship is removed, or when the connector shuts down. Free up any resources that this processor holds. The resource connector supplied by the base class is disconnected for you.

Processing events for a catalog target¶

If the catalog target processor implements OpenMetadataEventListener , it is passed each event from the open metadata ecosystem.

For this to happen, the integration connector must also implement OpenMetadataEventListener - this is the signal that the connector is interested in open metadata events. DynamicIntegrationConnectorBase provides the processEvent() implementation that fans the event out to the processors, and registers the listener at the end of the first refresh(). The delay is deliberate: it avoids a flood of events caused by the connector's own initial synchronization. The listener is only registered if the permitted synchronization allows metadata to flow to the third party technology.

public class MyIntegrationConnector extends DynamicIntegrationConnectorBase implements OpenMetadataEventListener
{
    // processEvent() is inherited from DynamicIntegrationConnectorBase
}

public class MyCatalogTargetProcessor extends CatalogTargetProcessorBase implements OpenMetadataEventListener
{
    @Override
    public void processEvent(OpenMetadataOutTopicEvent event)
    {
        // called for each open metadata event, for this catalog target
    }
}

The base class only passes events on when a refresh is not running, so your processEvent() method does not have to test integrationContext.noRefreshInProgress() itself.

Events from the third party technology are handled differently: register the listener in the processor's constructor or start() method. The OpenLineage Event Receiver Catalog Target Processor has an Open Metadata Topic Connector as its target resource connector, and registers a listener with it:

    public OpenLineageEventReceiverCatalogTargetProcessor(CatalogTarget             template,
                                                          CatalogTargetContext      catalogTargetContext,
                                                          Connector                 connectorToTarget,
                                                          String                    connectorName,
                                                          AuditLog                  auditLog,
                                                          OpenMetadataTopicListener listener) throws ConnectorCheckedException,
                                                                                                     UserNotAuthorizedException
    {
        super(template, catalogTargetContext, connectorToTarget, connectorName, auditLog);

        if (super.getConnectorToTarget() instanceof OpenMetadataTopicConnector topicConnector)
        {
            topicConnector.registerListener(listener);

            if (! topicConnector.isActive())
            {
                topicConnector.start();
            }
        }
    }

Its refresh() method does nothing, because all of its work is event-driven.

Being notified when the catalog targets change¶

An integration connector that needs to know when its catalog targets are added, changed or removed can implement CatalogTargetChangeListener and register it with the catalog targets manager:

        catalogTargetsManager.registerCatalogTargetChangeListener(this);

The interface has newCatalogTarget(), updatedCatalogTarget() and removedCatalogTarget() methods. Most connectors do not need this, since the processor's own start() and disconnect() methods cover the same events.

Declaring the supported catalog target types¶

The connector provider declares which types of element this connector accepts as a catalog target. This information is published in open metadata so that the tools people use to attach catalog targets can offer the right choices, and so that they can be validated. See defining the catalog target types.

Naming the metadata source¶

Each catalog target can name the software capability that acts as the metadata collection for the metadata created from it - the metadataSourceQualifiedName property on the CatalogTarget relationship. This means metadata from the sales database and metadata from the finance database can be held in separate metadata collections, even though a single connector instance created both.

CatalogTargetProcessorBase provides setUpMetadataSource() and setUpSoftwareCapability() to locate (or create) these elements. The external source metadata provenance that results prevents other tools from changing the metadata that this connector maintains.

Migrating an existing connector¶

To convert a connector that predates catalog targets:

1. Change the superclass from IntegrationConnectorBase (or one of the old XXXIntegratorConnector classes) to DynamicIntegrationConnectorBase.
2. Create a new class extending CatalogTargetProcessorBase and move the body of refresh() into it, along with any instance variables it uses.
3. Move the parts of start() that relate to a specific resource into the processor's start() method. Anything that reads the connector's own configuration stays behind.
4. Implement getNewRequestedCatalogTargetSkeleton() on the connector to return an instance of the new processor.
5. Replace getContext() with the integrationContext variable, and connectionProperties with connectionBean.
6. Replace direct use of the endpoint from the connector's connection with getConnectorToTarget(), or with the network address from the target asset's connection.
7. Declare the supported catalog target types in the connector provider.

Raise an issue or comment below