@Test public void shouldThrowExceptionIfApplicationServerConfigPortIsNotAnInteger() throws Exception { final Properties properties = configProps(); final String myEndPoint = "localhost:j87yhk"; properties.put(StreamsConfig.APPLICATION_SERVER_CONFIG, myEndPoint); final StreamsConfig config = new StreamsConfig(properties); final UUID uuid1 = UUID.randomUUID(); final String client1 = "client1"; final String applicationId = "application-id"; builder.setApplicationId(applicationId); final StreamThread streamThread = new StreamThread(builder, config, mockClientSupplier, applicationId, client1, uuid1, new Metrics(), Time.SYSTEM, new StreamsMetadataState(builder, StreamsMetadataState.UNKNOWN_HOST), 0); try { partitionAssignor.configure(config.getConsumerConfigs(streamThread, applicationId, client1)); Assert.fail("expected to an exception due to invalid config"); } catch (ConfigException e) { } }
@Override public void configure(Map<String, ?> configs) { numStandbyReplicas = (Integer) configs.get(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG); Object o = configs.get(StreamsConfig.InternalConfig.STREAM_THREAD_INSTANCE); if (o == null) { KafkaException ex = new KafkaException("StreamThread is not specified"); log.error(ex.getMessage(), ex); throw ex; } if (!(o instanceof StreamThread)) { KafkaException ex = new KafkaException(String.format("%s is not an instance of %s", o.getClass().getName(), StreamThread.class.getName())); log.error(ex.getMessage(), ex); throw ex; } streamThread = (StreamThread) o; streamThread.setPartitionAssignor(this); String userEndPoint = (String) configs.get(StreamsConfig.APPLICATION_SERVER_CONFIG); if (userEndPoint != null && !userEndPoint.isEmpty()) { try { String host = getHost(userEndPoint); Integer port = getPort(userEndPoint); if (host == null || port == null) throw new ConfigException(String.format("stream-thread [%s] Config %s isn't in the correct format. Expected a host:port pair" + " but received %s", streamThread.getName(), StreamsConfig.APPLICATION_SERVER_CONFIG, userEndPoint)); } catch (NumberFormatException nfe) { throw new ConfigException(String.format("stream-thread [%s] Invalid port supplied in %s for config %s", streamThread.getName(), userEndPoint, StreamsConfig.APPLICATION_SERVER_CONFIG)); } this.userEndPoint = userEndPoint; } internalTopicManager = new InternalTopicManager( new StreamsKafkaClient(this.streamThread.config), configs.containsKey(StreamsConfig.REPLICATION_FACTOR_CONFIG) ? (Integer) configs.get(StreamsConfig.REPLICATION_FACTOR_CONFIG) : 1, configs.containsKey(StreamsConfig.WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG) ? (Long) configs.get(StreamsConfig.WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG) : WINDOW_CHANGE_LOG_ADDITIONAL_RETENTION_DEFAULT, time); this.copartitionedTopicsValidator = new CopartitionedTopicsValidator(streamThread.getName()); }
StreamPartitionAssignor implements PartitionAssignor, Configurable { @Override public void configure(Map<String, ?> configs) { numStandbyReplicas = (Integer) configs.get(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG); Object o = configs.get(StreamsConfig.InternalConfig.STREAM_THREAD_INSTANCE); if (o == null) { KafkaException ex = new KafkaException("StreamThread is not specified"); log.error(ex.getMessage(), ex); throw ex; } if (!(o instanceof StreamThread)) { KafkaException ex = new KafkaException(String.format("%s is not an instance of %s", o.getClass().getName(), StreamThread.class.getName())); log.error(ex.getMessage(), ex); throw ex; } streamThread = (StreamThread) o; streamThread.setPartitionAssignor(this); String userEndPoint = (String) configs.get(StreamsConfig.APPLICATION_SERVER_CONFIG); if (userEndPoint != null && !userEndPoint.isEmpty()) { try { String host = getHost(userEndPoint); Integer port = getPort(userEndPoint); if (host == null || port == null) throw new ConfigException(String.format("stream-thread [%s] Config %s isn't in the correct format. Expected a host:port pair" + " but received %s", streamThread.getName(), StreamsConfig.APPLICATION_SERVER_CONFIG, userEndPoint)); } catch (NumberFormatException nfe) { throw new ConfigException(String.format("stream-thread [%s] Invalid port supplied in %s for config %s", streamThread.getName(), userEndPoint, StreamsConfig.APPLICATION_SERVER_CONFIG)); } this.userEndPoint = userEndPoint; } internalTopicManager = new InternalTopicManager( new StreamsKafkaClient(this.streamThread.config), configs.containsKey(StreamsConfig.REPLICATION_FACTOR_CONFIG) ? (Integer) configs.get(StreamsConfig.REPLICATION_FACTOR_CONFIG) : 1, configs.containsKey(StreamsConfig.WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG) ? (Long) configs.get(StreamsConfig.WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG) : WINDOW_CHANGE_LOG_ADDITIONAL_RETENTION_DEFAULT, time); this.copartitionedTopicsValidator = new CopartitionedTopicsValidator(streamThread.getName()); } }
StreamPartitionAssignor implements PartitionAssignor, Configurable { @Override public void configure(Map<String, ?> configs) { numStandbyReplicas = (Integer) configs.get(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG); Object o = configs.get(StreamsConfig.InternalConfig.STREAM_THREAD_INSTANCE); if (o == null) { KafkaException ex = new KafkaException("StreamThread is not specified"); log.error(ex.getMessage(), ex); throw ex; } if (!(o instanceof StreamThread)) { KafkaException ex = new KafkaException(String.format("%s is not an instance of %s", o.getClass().getName(), StreamThread.class.getName())); log.error(ex.getMessage(), ex); throw ex; } streamThread = (StreamThread) o; streamThread.setPartitionAssignor(this); String userEndPoint = (String) configs.get(StreamsConfig.APPLICATION_SERVER_CONFIG); if (userEndPoint != null && !userEndPoint.isEmpty()) { try { String host = getHost(userEndPoint); Integer port = getPort(userEndPoint); if (host == null || port == null) throw new ConfigException(String.format("stream-thread [%s] Config %s isn't in the correct format. Expected a host:port pair" + " but received %s", streamThread.getName(), StreamsConfig.APPLICATION_SERVER_CONFIG, userEndPoint)); } catch (NumberFormatException nfe) { throw new ConfigException(String.format("stream-thread [%s] Invalid port supplied in %s for config %s", streamThread.getName(), userEndPoint, StreamsConfig.APPLICATION_SERVER_CONFIG)); } this.userEndPoint = userEndPoint; } internalTopicManager = new InternalTopicManager( new StreamsKafkaClient(this.streamThread.config), configs.containsKey(StreamsConfig.REPLICATION_FACTOR_CONFIG) ? (Integer) configs.get(StreamsConfig.REPLICATION_FACTOR_CONFIG) : 1, configs.containsKey(StreamsConfig.WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG) ? (Long) configs.get(StreamsConfig.WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG) : WINDOW_CHANGE_LOG_ADDITIONAL_RETENTION_DEFAULT, time); this.copartitionedTopicsValidator = new CopartitionedTopicsValidator(streamThread.getName()); }  }
StreamPartitionAssignor implements PartitionAssignor, Configurable { @Override public void configure(Map<String, ?> configs) { numStandbyReplicas = (Integer) configs.get(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG); Object o = configs.get(StreamsConfig.InternalConfig.STREAM_THREAD_INSTANCE); if (o == null) { KafkaException ex = new KafkaException("StreamThread is not specified"); log.error(ex.getMessage(), ex); throw ex; } if (!(o instanceof StreamThread)) { KafkaException ex = new KafkaException(String.format("%s is not an instance of %s", o.getClass().getName(), StreamThread.class.getName())); log.error(ex.getMessage(), ex); throw ex; } streamThread = (StreamThread) o; streamThread.setPartitionAssignor(this); String userEndPoint = (String) configs.get(StreamsConfig.APPLICATION_SERVER_CONFIG); if (userEndPoint != null && !userEndPoint.isEmpty()) { try { String host = getHost(userEndPoint); Integer port = getPort(userEndPoint); if (host == null || port == null) throw new ConfigException(String.format("stream-thread [%s] Config %s isn't in the correct format. Expected a host:port pair" + " but received %s", streamThread.getName(), StreamsConfig.APPLICATION_SERVER_CONFIG, userEndPoint)); } catch (NumberFormatException nfe) { throw new ConfigException(String.format("stream-thread [%s] Invalid port supplied in %s for config %s", streamThread.getName(), userEndPoint, StreamsConfig.APPLICATION_SERVER_CONFIG)); } this.userEndPoint = userEndPoint; } internalTopicManager = new InternalTopicManager( new StreamsKafkaClient(this.streamThread.config), configs.containsKey(StreamsConfig.REPLICATION_FACTOR_CONFIG) ? (Integer) configs.get(StreamsConfig.REPLICATION_FACTOR_CONFIG) : 1, configs.containsKey(StreamsConfig.WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG) ? (Long) configs.get(StreamsConfig.WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG) : WINDOW_CHANGE_LOG_ADDITIONAL_RETENTION_DEFAULT, time); this.copartitionedTopicsValidator = new CopartitionedTopicsValidator(streamThread.getName()); }  @Override void configure(Map<String, ?> configs); @Override String name(); @Override Subscription subscription(Set<String> topics); @Override Map<String, Assignment> assign(Cluster metadata, Map<String, Subscription> subscriptions); @Override void onAssignment(Assignment assignment); void close(); }
StreamPartitionAssignor implements PartitionAssignor, Configurable { @Override public void configure(Map<String, ?> configs) { numStandbyReplicas = (Integer) configs.get(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG); Object o = configs.get(StreamsConfig.InternalConfig.STREAM_THREAD_INSTANCE); if (o == null) { KafkaException ex = new KafkaException("StreamThread is not specified"); log.error(ex.getMessage(), ex); throw ex; } if (!(o instanceof StreamThread)) { KafkaException ex = new KafkaException(String.format("%s is not an instance of %s", o.getClass().getName(), StreamThread.class.getName())); log.error(ex.getMessage(), ex); throw ex; } streamThread = (StreamThread) o; streamThread.setPartitionAssignor(this); String userEndPoint = (String) configs.get(StreamsConfig.APPLICATION_SERVER_CONFIG); if (userEndPoint != null && !userEndPoint.isEmpty()) { try { String host = getHost(userEndPoint); Integer port = getPort(userEndPoint); if (host == null || port == null) throw new ConfigException(String.format("stream-thread [%s] Config %s isn't in the correct format. Expected a host:port pair" + " but received %s", streamThread.getName(), StreamsConfig.APPLICATION_SERVER_CONFIG, userEndPoint)); } catch (NumberFormatException nfe) { throw new ConfigException(String.format("stream-thread [%s] Invalid port supplied in %s for config %s", streamThread.getName(), userEndPoint, StreamsConfig.APPLICATION_SERVER_CONFIG)); } this.userEndPoint = userEndPoint; } internalTopicManager = new InternalTopicManager( new StreamsKafkaClient(this.streamThread.config), configs.containsKey(StreamsConfig.REPLICATION_FACTOR_CONFIG) ? (Integer) configs.get(StreamsConfig.REPLICATION_FACTOR_CONFIG) : 1, configs.containsKey(StreamsConfig.WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG) ? (Long) configs.get(StreamsConfig.WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG) : WINDOW_CHANGE_LOG_ADDITIONAL_RETENTION_DEFAULT, time); this.copartitionedTopicsValidator = new CopartitionedTopicsValidator(streamThread.getName()); }  @Override void configure(Map<String, ?> configs); @Override String name(); @Override Subscription subscription(Set<String> topics); @Override Map<String, Assignment> assign(Cluster metadata, Map<String, Subscription> subscriptions); @Override void onAssignment(Assignment assignment); void close(); final static int NOT_AVAILABLE; }