@Test public void testBrokersToEndpoints() { List<Broker> brokersList = new ArrayList<Broker>(4); brokersList .add(new Broker(0, "localhost", 1, new ListenerName("CLIENT"), SecurityProtocol.PLAINTEXT)); brokersList.add(new Broker(1, "localhost1", 12, ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT), SecurityProtocol.PLAINTEXT )); brokersList.add(new Broker(2, "localhost2", 123, new ListenerName("SECURE_REPLICATION"), SecurityProtocol.SASL_PLAINTEXT )); brokersList.add(new Broker(2, "localhost2", 123, ListenerName.forSecurityProtocol(SecurityProtocol.SASL_PLAINTEXT), SecurityProtocol.SASL_PLAINTEXT )); brokersList.add(new Broker(3, "localhost3", 1234, ListenerName.forSecurityProtocol(SecurityProtocol.SSL), SecurityProtocol.SSL )); List<String> endpointsList = SchemaRegistryConfig.brokersToEndpoints((brokersList)); List<String> expected = new ArrayList<String>(4); expected.add("PLAINTEXT: expected.add("PLAINTEXT: expected.add("SASL_PLAINTEXT: expected.add("SASL_PLAINTEXT: expected.add("SSL: assertEquals("Expected the same size list.", expected.size(), endpointsList.size()); for (int i = 0; i < endpointsList.size(); i++) { assertEquals("Expected a different endpoint", expected.get(i), endpointsList.get(i)); } }
static List<String> brokersToEndpoints(List<Broker> brokers) { final List<String> endpoints = new LinkedList<>(); for (Broker broker : brokers) { for (EndPoint ep : CollectionConverters.asJavaCollection(broker.endPoints())) { String hostport = ep.host() == null ? ":" + ep.port() : Utils.formatAddress(ep.host(), ep.port()); String endpoint = ep.securityProtocol() + ": endpoints.add(endpoint); } } return endpoints; }
SchemaRegistryConfig extends RestConfig { static List<String> brokersToEndpoints(List<Broker> brokers) { final List<String> endpoints = new LinkedList<>(); for (Broker broker : brokers) { for (EndPoint ep : CollectionConverters.asJavaCollection(broker.endPoints())) { String hostport = ep.host() == null ? ":" + ep.port() : Utils.formatAddress(ep.host(), ep.port()); String endpoint = ep.securityProtocol() + ": endpoints.add(endpoint); } } return endpoints; } }
SchemaRegistryConfig extends RestConfig { static List<String> brokersToEndpoints(List<Broker> brokers) { final List<String> endpoints = new LinkedList<>(); for (Broker broker : brokers) { for (EndPoint ep : CollectionConverters.asJavaCollection(broker.endPoints())) { String hostport = ep.host() == null ? ":" + ep.port() : Utils.formatAddress(ep.host(), ep.port()); String endpoint = ep.securityProtocol() + ": endpoints.add(endpoint); } } return endpoints; } SchemaRegistryConfig(String propsFile);  SchemaRegistryConfig(Properties props);  SchemaRegistryConfig(ConfigDef configDef, Properties props); }
SchemaRegistryConfig extends RestConfig { static List<String> brokersToEndpoints(List<Broker> brokers) { final List<String> endpoints = new LinkedList<>(); for (Broker broker : brokers) { for (EndPoint ep : CollectionConverters.asJavaCollection(broker.endPoints())) { String hostport = ep.host() == null ? ":" + ep.port() : Utils.formatAddress(ep.host(), ep.port()); String endpoint = ep.securityProtocol() + ": endpoints.add(endpoint); } } return endpoints; } SchemaRegistryConfig(String propsFile);  SchemaRegistryConfig(Properties props);  SchemaRegistryConfig(ConfigDef configDef, Properties props); static ConfigDef baseSchemaRegistryConfigDef(); Properties originalProperties(); CompatibilityLevel compatibilityType(); boolean useKafkaCoordination(); String bootstrapBrokers(); boolean checkZkAclConfig(); String definedResourceExtensionConfigName(); List<String> getStaticLocations(); String interInstanceProtocol(); synchronized ZkUtils zkUtils(); List<String> whitelistHeaders(); static void main(String[] args); }
SchemaRegistryConfig extends RestConfig { static List<String> brokersToEndpoints(List<Broker> brokers) { final List<String> endpoints = new LinkedList<>(); for (Broker broker : brokers) { for (EndPoint ep : CollectionConverters.asJavaCollection(broker.endPoints())) { String hostport = ep.host() == null ? ":" + ep.port() : Utils.formatAddress(ep.host(), ep.port()); String endpoint = ep.securityProtocol() + ": endpoints.add(endpoint); } } return endpoints; } SchemaRegistryConfig(String propsFile);  SchemaRegistryConfig(Properties props);  SchemaRegistryConfig(ConfigDef configDef, Properties props); static ConfigDef baseSchemaRegistryConfigDef(); Properties originalProperties(); CompatibilityLevel compatibilityType(); boolean useKafkaCoordination(); String bootstrapBrokers(); boolean checkZkAclConfig(); String definedResourceExtensionConfigName(); List<String> getStaticLocations(); String interInstanceProtocol(); synchronized ZkUtils zkUtils(); List<String> whitelistHeaders(); static void main(String[] args); static final String SCHEMAREGISTRY_GROUP_ID_CONFIG; @Deprecated
  static final String KAFKASTORE_SECURITY_PROTOCOL_SSL; @Deprecated
  static final String KAFKASTORE_SECURITY_PROTOCOL_PLAINTEXT; @Deprecated
  static final String KAFKASTORE_CONNECTION_URL_CONFIG; static final String KAFKASTORE_BOOTSTRAP_SERVERS_CONFIG; static final String KAFKASTORE_GROUP_ID_CONFIG; static final String KAFKASTORE_ZK_SESSION_TIMEOUT_MS_CONFIG; static final String KAFKASTORE_TOPIC_CONFIG; static final String DEFAULT_KAFKASTORE_TOPIC; static final String KAFKASTORE_TOPIC_REPLICATION_FACTOR_CONFIG; static final int DEFAULT_KAFKASTORE_TOPIC_REPLICATION_FACTOR; static final String KAFKASTORE_WRITE_MAX_RETRIES_CONFIG; static final int DEFAULT_KAFKASTORE_WRITE_MAX_RETRIES; static final String KAFKASTORE_TIMEOUT_CONFIG; static final String KAFKASTORE_CHECKPOINT_DIR_CONFIG; static final String KAFKASTORE_CHECKPOINT_VERSION_CONFIG; static final String KAFKASTORE_INIT_TIMEOUT_CONFIG; static final String KAFKASTORE_UPDATE_HANDLERS_CONFIG; @Deprecated
  static final String MASTER_ELIGIBILITY; static final String LEADER_ELIGIBILITY; static final boolean DEFAULT_LEADER_ELIGIBILITY; static final String MODE_MUTABILITY; static final boolean DEFAULT_MODE_MUTABILITY; static final String SCHEMAREGISTRY_ZK_NAMESPACE; static final String DEFAULT_SCHEMAREGISTRY_ZK_NAMESPACE; static final String HOST_NAME_CONFIG; static final String SCHEMA_PROVIDERS_CONFIG; @Deprecated
  static final String COMPATIBILITY_CONFIG; static final String SCHEMA_COMPATIBILITY_CONFIG; static final String ZOOKEEPER_SET_ACL_CONFIG; static final String KAFKASTORE_SECURITY_PROTOCOL_CONFIG; static final String KAFKASTORE_SSL_TRUSTSTORE_LOCATION_CONFIG; static final String KAFKASTORE_SSL_TRUSTSTORE_PASSWORD_CONFIG; static final String KAFKASTORE_SSL_KEYSTORE_LOCATION_CONFIG; static final String KAFKASTORE_SSL_TRUSTSTORE_TYPE_CONFIG; static final String KAFKASTORE_SSL_TRUSTMANAGER_ALGORITHM_CONFIG; static final String KAFKASTORE_SSL_KEYSTORE_PASSWORD_CONFIG; static final String KAFKASTORE_SSL_KEYSTORE_TYPE_CONFIG; static final String KAFKASTORE_SSL_KEYMANAGER_ALGORITHM_CONFIG; static final String KAFKASTORE_SSL_KEY_PASSWORD_CONFIG; static final String KAFKASTORE_SSL_ENABLED_PROTOCOLS_CONFIG; static final String KAFKASTORE_SSL_PROTOCOL_CONFIG; static final String KAFKASTORE_SSL_PROVIDER_CONFIG; static final String KAFKASTORE_SSL_CIPHER_SUITES_CONFIG; static final String KAFKASTORE_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG; static final String KAFKASTORE_SASL_KERBEROS_SERVICE_NAME_CONFIG; static final String KAFKASTORE_SASL_MECHANISM_CONFIG; static final String KAFKASTORE_SASL_KERBEROS_KINIT_CMD_CONFIG; static final String KAFKASTORE_SASL_KERBEROS_MIN_TIME_BEFORE_RELOGIN_CONFIG; static final String KAFKASTORE_SASL_KERBEROS_TICKET_RENEW_JITTER_CONFIG; static final String KAFKASTORE_SASL_KERBEROS_TICKET_RENEW_WINDOW_FACTOR_CONFIG; @Deprecated
  static final String SCHEMAREGISTRY_RESOURCE_EXTENSION_CONFIG; static final String RESOURCE_EXTENSION_CONFIG; static final String RESOURCE_STATIC_LOCATIONS_CONFIG; @Deprecated
  static final String SCHEMAREGISTRY_INTER_INSTANCE_PROTOCOL_CONFIG; static final String INTER_INSTANCE_PROTOCOL_CONFIG; static final String INTER_INSTANCE_HEADERS_WHITELIST_CONFIG; static final String KAFKASTORE_SASL_KERBEROS_SERVICE_NAME_DOC; static final String KAFKASTORE_SASL_MECHANISM_DOC; static final String KAFKASTORE_SASL_KERBEROS_KINIT_CMD_DOC; static final String KAFKASTORE_SASL_KERBEROS_MIN_TIME_BEFORE_RELOGIN_DOC; static final String KAFKASTORE_SASL_KERBEROS_TICKET_RENEW_JITTER_DOC; static final String KAFKASTORE_SASL_KERBEROS_TICKET_RENEW_WINDOW_FACTOR_DOC; static final String HTTPS; static final String HTTP; }