@Override
    public Consumer createConsumer(Processor processor) throws Exception {
        // we need to have database and container set to consume events
        if (ObjectHelper.isEmpty(configuration.getDatabaseName()) || ObjectHelper.isEmpty(configuration.getContainerName())) {
            throw new IllegalArgumentException("Database name and container name must be set.");
        }

        final Consumer cosmosDbConsumer = new CosmosDbConsumer(this, processor);
        configureConsumer(cosmosDbConsumer);

        return cosmosDbConsumer;
    }
@Test
    void testCreateConsumerWithInvalidConfig() throws Exception {
        final String uri = "azure-cosmosdb://mydb/myContainer";
        String remaining = "mydb";
        final Map<String, Object> params = new HashMap<>();
        params.put("databaseEndpoint", "https://test.com:443");
        params.put("createDatabaseIfNotExists", "true");
        params.put("accountKey", "myKey");

        final CosmosDbEndpoint endpoint = (CosmosDbEndpoint) context.getComponent("azure-cosmosdb", CosmosDbComponent.class)
                .createEndpoint(uri, remaining, params);
        assertThrows(IllegalArgumentException.class, () -> endpoint.createConsumer(exchange -> {
        }));

        params.put("databaseEndpoint", "https://test.com:443");
        params.put("createDatabaseIfNotExists", "true");
        params.put("accountKey", "myKey");
        remaining = "/mydb";
        final CosmosDbEndpoint endpoint2 = (CosmosDbEndpoint) context.getComponent("azure-cosmosdb", CosmosDbComponent.class)
                .createEndpoint(uri, remaining, params);
        assertThrows(IllegalArgumentException.class, () -> endpoint2.createConsumer(exchange -> {
        }));

        params.put("databaseEndpoint", "https://test.com:443");
        params.put("createDatabaseIfNotExists", "true");
        params.put("accountKey", "myKey");
        remaining = "mydb/";
        final CosmosDbEndpoint endpoint3 = (CosmosDbEndpoint) context.getComponent("azure-cosmosdb", CosmosDbComponent.class)
                .createEndpoint(uri, remaining, params);
        assertThrows(IllegalArgumentException.class, () -> endpoint3.createConsumer(exchange -> {
        }));

        params.put("databaseEndpoint", "https://test.com:443");
        params.put("createDatabaseIfNotExists", "true");
        params.put("accountKey", "myKey");
        remaining = "";
        final CosmosDbEndpoint endpoint4 = (CosmosDbEndpoint) context.getComponent("azure-cosmosdb", CosmosDbComponent.class)
                .createEndpoint(uri, remaining, params);
        assertThrows(IllegalArgumentException.class, () -> endpoint4.createConsumer(exchange -> {
        }));

        params.put("databaseEndpoint", "https://test.com:443");
        params.put("createDatabaseIfNotExists", "true");
        params.put("accountKey", "myKey");
        remaining = "mydb/mycn";
        final CosmosDbEndpoint endpoint5 = (CosmosDbEndpoint) context.getComponent("azure-cosmosdb", CosmosDbComponent.class)
                .createEndpoint(uri, remaining, params);
        assertNotNull(endpoint5.createConsumer(exchange -> {
        }));
    }