@Override
    protected Endpoint createEndpoint(final String uri, final String path, final Map<String, Object> parameters)
            throws Exception {
        if (autoConfiguration != null && autoConfiguration.isAutoConfigurable()) {
            autoConfiguration.ensureNameSpaceAndTenant(path);
        }

        final PulsarConfiguration copy = configuration.copy();

        PulsarEndpoint answer = new PulsarEndpoint(uri, this);
        answer.setPulsarConfiguration(copy);
        answer.setPulsarClient(pulsarClient);
        setProperties(answer, parameters);

        PulsarPath pp = new PulsarPath(path);
        if (pp.isAutoConfigurable()) {
            answer.setPersistence(pp.getPersistence());
            answer.setTenant(pp.getTenant());
            answer.setNamespace(pp.getNamespace());
            answer.setTopic(pp.getTopic());
        } else {
            throw new IllegalArgumentException("Pulsar name structure is invalid: was " + path);
        }

        return answer;
    }
@Test
    public void testPulsarEndpointDefaultConfiguration() throws Exception {
        PulsarComponent component = context.getComponent("pulsar", PulsarComponent.class);

        PulsarEndpoint endpoint = (PulsarEndpoint) component.createEndpoint("pulsar://persistent/test/foobar/BatchCreated");

        assertNotNull(endpoint);
        assertEquals("sole-consumer", endpoint.getPulsarConfiguration().getConsumerName());
        assertEquals("cons", endpoint.getPulsarConfiguration().getConsumerNamePrefix());
        assertEquals(10, endpoint.getPulsarConfiguration().getConsumerQueueSize());
        assertEquals(1, endpoint.getPulsarConfiguration().getNumberOfConsumers());
        assertNull(endpoint.getPulsarConfiguration().getProducerName());
        assertEquals("subs", endpoint.getPulsarConfiguration().getSubscriptionName());
        assertEquals(SubscriptionType.EXCLUSIVE, endpoint.getPulsarConfiguration().getSubscriptionType());
        assertFalse(endpoint.getPulsarConfiguration().isAllowManualAcknowledgement());
        assertFalse(endpoint.getPulsarConfiguration().isReadCompacted());
        assertTrue(endpoint.getPulsarConfiguration().isMessageListener());
    }