Brooklin does not alllow multiple datastreams replicating to the same destination Kafka topic

[miguelmduarte]

Subject of the issue

Brooklin does not allow creating multiple datastreams replicating from different source Kafka clusters/topics to the same destination Kafka cluster and topic.

Let's say we have the following Kafka clusters: K1, K2

We wish to replicate messages from topic T1 in K1 and K2 clusters to topic T2 in K1 cluster.

We manage to create the first datastream (K2-T1 -> K1-T2) but the creation of the second one (K1-T1 -> K1-T2) fails with the error:

Cannot create a BYOT datastream where the destination is being used by other datastream(s)

Your environment

• Operating System: 3-node brooklin cluster running in centOS-7 VMs
• Brooklin version: Compiling and deploying the brooklin source code currently in master
• Java version: openjdk 8u272+b10
• Kafka version: 2.12_2.5.1
• ZooKeeper version: 3.6.2

Steps to reproduce

1. Create the first DS:

{
    "name": "my-datastream-1",
    "connectorName": "kafkaConnector",
    "transportProviderName": "kafkaTransportProvider",
    "source": {
        "connectionString": "kafka://K2-broker001:9092,K2-broker002:9092,K2-broker003:9092/T1"
    },
    "destination": {
        "connectionString": "kafka://K1-broker001:9092,K1-broker002:9092,K1-broker003:9092/T2"
    },
    "metadata": {
        "group.id": "my_group",
        "system.auto.offset.reset": "latest",
        "system.reuseExistingDestination": "true",
        "system.IsUserManagedDestination": "false",
        "system.IsConnectorManagedDestination": "true"
    }
}

Datastream is created successfully.

2. Create the second DS:

{
    "name": "my-datastream-2",
    "connectorName": "kafkaConnector",
    "transportProviderName": "kafkaTransportProvider",
    "source": {
        "connectionString": "kafka://K1-broker001:9092,K1-broker002:9092,K1-broker003:9092/T1"
    },
    "destination": {
        "connectionString": "kafka://K1-broker001:9092,K1-broker002:9092,K1-broker003:9092/T2"
    },
    "metadata": {
        "group.id": "my_group",
        "system.auto.offset.reset": "latest",
        "system.reuseExistingDestination": "true",
        "system.IsUserManagedDestination": "false",
        "system.IsConnectorManagedDestination": "true"
    }
}

We get the following error:

Cannot create a BYOT datastream where the destination is being used by other datastream(s)

3. If we swap the broker order in the destination connection string of the second DS, we manage to create the DS successfully.

{
    "name": "my-datastream-2",
    "connectorName": "kafkaConnector",
    "transportProviderName": "kafkaTransportProvider",
    "source": {
        "connectionString": "kafka://K1-broker001:9092,K1-broker002:9092,K1-broker003:9092/T1"
    },
    "destination": {
        "connectionString": "kafka://K1-broker002:9092,K1-broker001:9092,K1-broker003:9092/T2"
    },
    "metadata": {
        "group.id": "my_group",
        "system.auto.offset.reset": "latest",
        "system.reuseExistingDestination": "true",
        "system.IsUserManagedDestination": "false",
        "system.IsConnectorManagedDestination": "true"
    }
}

Expected behaviour

We expected that it would be possible to replicate from multiple source Kafka clusters/topics to the same Kafka destination cluster and topic.

Actual behaviour

The second DS fails when another DS with the same destination already exists. If we swap the broker order in the second DS destination connection string, we manage to successfully create the second DS.