@Test public void testTimeoutWithoutMetadata() throws Exception { try (MockKafkaAdminClientEnv env = mockClientEnv(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, "10")) { env.kafkaClient().setNodeApiVersions(NodeApiVersions.create()); env.kafkaClient().setNode(new Node(0, "localhost", 8121)); env.kafkaClient().prepareResponse(new CreateTopicsResponse(Collections.singletonMap("myTopic", new ApiError(Errors.NONE, "")))); KafkaFuture<Void> future = env.adminClient().createTopics( Collections.singleton(new NewTopic("myTopic", Collections.singletonMap(Integer.valueOf(0), asList(new Integer[]{0, 1, 2})))), new CreateTopicsOptions().timeoutMs(1000)).all(); assertFutureError(future, TimeoutException.class); } }
@Override public CreateTopicsResult createTopics(final Collection<NewTopic> newTopics, final CreateTopicsOptions options) { final Map<String, KafkaFutureImpl<Void>> topicFutures = new HashMap<>(newTopics.size()); final Map<String, CreateTopicsRequest.TopicDetails> topicsMap = new HashMap<>(newTopics.size()); for (NewTopic newTopic : newTopics) { if (topicFutures.get(newTopic.name()) == null) { topicFutures.put(newTopic.name(), new KafkaFutureImpl<Void>()); topicsMap.put(newTopic.name(), newTopic.convertToTopicDetails()); } } final long now = time.milliseconds(); runnable.call(new Call("createTopics", calcDeadlineMs(now, options.timeoutMs()), new ControllerNodeProvider()) { @Override public AbstractRequest.Builder createRequest(int timeoutMs) { return new CreateTopicsRequest.Builder(topicsMap, timeoutMs, options.shouldValidateOnly()); } @Override public void handleResponse(AbstractResponse abstractResponse) { CreateTopicsResponse response = (CreateTopicsResponse) abstractResponse; for (Map.Entry<String, ApiError> entry : response.errors().entrySet()) { KafkaFutureImpl<Void> future = topicFutures.get(entry.getKey()); if (future == null) { log.warn("Server response mentioned unknown topic {}", entry.getKey()); } else { ApiException exception = entry.getValue().exception(); if (exception != null) { future.completeExceptionally(exception); } else { future.complete(null); } } } for (Map.Entry<String, KafkaFutureImpl<Void>> entry : topicFutures.entrySet()) { KafkaFutureImpl<Void> future = entry.getValue(); if (!future.isDone()) { future.completeExceptionally(new ApiException("The server response did not " + "contain a reference to node " + entry.getKey())); } } } @Override void handleFailure(Throwable throwable) { completeAllExceptionally(topicFutures.values(), throwable); } }, now); return new CreateTopicsResult(new HashMap<String, KafkaFuture<Void>>(topicFutures)); }
KafkaAdminClient extends AdminClient { @Override public CreateTopicsResult createTopics(final Collection<NewTopic> newTopics, final CreateTopicsOptions options) { final Map<String, KafkaFutureImpl<Void>> topicFutures = new HashMap<>(newTopics.size()); final Map<String, CreateTopicsRequest.TopicDetails> topicsMap = new HashMap<>(newTopics.size()); for (NewTopic newTopic : newTopics) { if (topicFutures.get(newTopic.name()) == null) { topicFutures.put(newTopic.name(), new KafkaFutureImpl<Void>()); topicsMap.put(newTopic.name(), newTopic.convertToTopicDetails()); } } final long now = time.milliseconds(); runnable.call(new Call("createTopics", calcDeadlineMs(now, options.timeoutMs()), new ControllerNodeProvider()) { @Override public AbstractRequest.Builder createRequest(int timeoutMs) { return new CreateTopicsRequest.Builder(topicsMap, timeoutMs, options.shouldValidateOnly()); } @Override public void handleResponse(AbstractResponse abstractResponse) { CreateTopicsResponse response = (CreateTopicsResponse) abstractResponse; for (Map.Entry<String, ApiError> entry : response.errors().entrySet()) { KafkaFutureImpl<Void> future = topicFutures.get(entry.getKey()); if (future == null) { log.warn("Server response mentioned unknown topic {}", entry.getKey()); } else { ApiException exception = entry.getValue().exception(); if (exception != null) { future.completeExceptionally(exception); } else { future.complete(null); } } } for (Map.Entry<String, KafkaFutureImpl<Void>> entry : topicFutures.entrySet()) { KafkaFutureImpl<Void> future = entry.getValue(); if (!future.isDone()) { future.completeExceptionally(new ApiException("The server response did not " + "contain a reference to node " + entry.getKey())); } } } @Override void handleFailure(Throwable throwable) { completeAllExceptionally(topicFutures.values(), throwable); } }, now); return new CreateTopicsResult(new HashMap<String, KafkaFuture<Void>>(topicFutures)); } }
KafkaAdminClient extends AdminClient { @Override public CreateTopicsResult createTopics(final Collection<NewTopic> newTopics, final CreateTopicsOptions options) { final Map<String, KafkaFutureImpl<Void>> topicFutures = new HashMap<>(newTopics.size()); final Map<String, CreateTopicsRequest.TopicDetails> topicsMap = new HashMap<>(newTopics.size()); for (NewTopic newTopic : newTopics) { if (topicFutures.get(newTopic.name()) == null) { topicFutures.put(newTopic.name(), new KafkaFutureImpl<Void>()); topicsMap.put(newTopic.name(), newTopic.convertToTopicDetails()); } } final long now = time.milliseconds(); runnable.call(new Call("createTopics", calcDeadlineMs(now, options.timeoutMs()), new ControllerNodeProvider()) { @Override public AbstractRequest.Builder createRequest(int timeoutMs) { return new CreateTopicsRequest.Builder(topicsMap, timeoutMs, options.shouldValidateOnly()); } @Override public void handleResponse(AbstractResponse abstractResponse) { CreateTopicsResponse response = (CreateTopicsResponse) abstractResponse; for (Map.Entry<String, ApiError> entry : response.errors().entrySet()) { KafkaFutureImpl<Void> future = topicFutures.get(entry.getKey()); if (future == null) { log.warn("Server response mentioned unknown topic {}", entry.getKey()); } else { ApiException exception = entry.getValue().exception(); if (exception != null) { future.completeExceptionally(exception); } else { future.complete(null); } } } for (Map.Entry<String, KafkaFutureImpl<Void>> entry : topicFutures.entrySet()) { KafkaFutureImpl<Void> future = entry.getValue(); if (!future.isDone()) { future.completeExceptionally(new ApiException("The server response did not " + "contain a reference to node " + entry.getKey())); } } } @Override void handleFailure(Throwable throwable) { completeAllExceptionally(topicFutures.values(), throwable); } }, now); return new CreateTopicsResult(new HashMap<String, KafkaFuture<Void>>(topicFutures)); } private  KafkaAdminClient(AdminClientConfig config, String clientId, Time time, Metadata metadata,
                     Metrics metrics, KafkaClient client, TimeoutProcessorFactory timeoutProcessorFactory); }
KafkaAdminClient extends AdminClient { @Override public CreateTopicsResult createTopics(final Collection<NewTopic> newTopics, final CreateTopicsOptions options) { final Map<String, KafkaFutureImpl<Void>> topicFutures = new HashMap<>(newTopics.size()); final Map<String, CreateTopicsRequest.TopicDetails> topicsMap = new HashMap<>(newTopics.size()); for (NewTopic newTopic : newTopics) { if (topicFutures.get(newTopic.name()) == null) { topicFutures.put(newTopic.name(), new KafkaFutureImpl<Void>()); topicsMap.put(newTopic.name(), newTopic.convertToTopicDetails()); } } final long now = time.milliseconds(); runnable.call(new Call("createTopics", calcDeadlineMs(now, options.timeoutMs()), new ControllerNodeProvider()) { @Override public AbstractRequest.Builder createRequest(int timeoutMs) { return new CreateTopicsRequest.Builder(topicsMap, timeoutMs, options.shouldValidateOnly()); } @Override public void handleResponse(AbstractResponse abstractResponse) { CreateTopicsResponse response = (CreateTopicsResponse) abstractResponse; for (Map.Entry<String, ApiError> entry : response.errors().entrySet()) { KafkaFutureImpl<Void> future = topicFutures.get(entry.getKey()); if (future == null) { log.warn("Server response mentioned unknown topic {}", entry.getKey()); } else { ApiException exception = entry.getValue().exception(); if (exception != null) { future.completeExceptionally(exception); } else { future.complete(null); } } } for (Map.Entry<String, KafkaFutureImpl<Void>> entry : topicFutures.entrySet()) { KafkaFutureImpl<Void> future = entry.getValue(); if (!future.isDone()) { future.completeExceptionally(new ApiException("The server response did not " + "contain a reference to node " + entry.getKey())); } } } @Override void handleFailure(Throwable throwable) { completeAllExceptionally(topicFutures.values(), throwable); } }, now); return new CreateTopicsResult(new HashMap<String, KafkaFuture<Void>>(topicFutures)); } private  KafkaAdminClient(AdminClientConfig config, String clientId, Time time, Metadata metadata,
                     Metrics metrics, KafkaClient client, TimeoutProcessorFactory timeoutProcessorFactory); @Override void close(long duration, TimeUnit unit); @Override CreateTopicsResult createTopics(final Collection<NewTopic> newTopics,
                                           final CreateTopicsOptions options); @Override DeleteTopicsResult deleteTopics(final Collection<String> topicNames,
                                           DeleteTopicsOptions options); @Override ListTopicsResult listTopics(final ListTopicsOptions options); @Override DescribeTopicsResult describeTopics(final Collection<String> topicNames, DescribeTopicsOptions options); @Override DescribeClusterResult describeCluster(DescribeClusterOptions options); @Override DescribeAclsResult describeAcls(final AclBindingFilter filter, DescribeAclsOptions options); @Override CreateAclsResult createAcls(Collection<AclBinding> acls, CreateAclsOptions options); @Override DeleteAclsResult deleteAcls(Collection<AclBindingFilter> filters, DeleteAclsOptions options); @Override DescribeConfigsResult describeConfigs(Collection<ConfigResource> configResources, final DescribeConfigsOptions options); @Override AlterConfigsResult alterConfigs(Map<ConfigResource, Config> configs, final AlterConfigsOptions options); }
KafkaAdminClient extends AdminClient { @Override public CreateTopicsResult createTopics(final Collection<NewTopic> newTopics, final CreateTopicsOptions options) { final Map<String, KafkaFutureImpl<Void>> topicFutures = new HashMap<>(newTopics.size()); final Map<String, CreateTopicsRequest.TopicDetails> topicsMap = new HashMap<>(newTopics.size()); for (NewTopic newTopic : newTopics) { if (topicFutures.get(newTopic.name()) == null) { topicFutures.put(newTopic.name(), new KafkaFutureImpl<Void>()); topicsMap.put(newTopic.name(), newTopic.convertToTopicDetails()); } } final long now = time.milliseconds(); runnable.call(new Call("createTopics", calcDeadlineMs(now, options.timeoutMs()), new ControllerNodeProvider()) { @Override public AbstractRequest.Builder createRequest(int timeoutMs) { return new CreateTopicsRequest.Builder(topicsMap, timeoutMs, options.shouldValidateOnly()); } @Override public void handleResponse(AbstractResponse abstractResponse) { CreateTopicsResponse response = (CreateTopicsResponse) abstractResponse; for (Map.Entry<String, ApiError> entry : response.errors().entrySet()) { KafkaFutureImpl<Void> future = topicFutures.get(entry.getKey()); if (future == null) { log.warn("Server response mentioned unknown topic {}", entry.getKey()); } else { ApiException exception = entry.getValue().exception(); if (exception != null) { future.completeExceptionally(exception); } else { future.complete(null); } } } for (Map.Entry<String, KafkaFutureImpl<Void>> entry : topicFutures.entrySet()) { KafkaFutureImpl<Void> future = entry.getValue(); if (!future.isDone()) { future.completeExceptionally(new ApiException("The server response did not " + "contain a reference to node " + entry.getKey())); } } } @Override void handleFailure(Throwable throwable) { completeAllExceptionally(topicFutures.values(), throwable); } }, now); return new CreateTopicsResult(new HashMap<String, KafkaFuture<Void>>(topicFutures)); } private  KafkaAdminClient(AdminClientConfig config, String clientId, Time time, Metadata metadata,
                     Metrics metrics, KafkaClient client, TimeoutProcessorFactory timeoutProcessorFactory); @Override void close(long duration, TimeUnit unit); @Override CreateTopicsResult createTopics(final Collection<NewTopic> newTopics,
                                           final CreateTopicsOptions options); @Override DeleteTopicsResult deleteTopics(final Collection<String> topicNames,
                                           DeleteTopicsOptions options); @Override ListTopicsResult listTopics(final ListTopicsOptions options); @Override DescribeTopicsResult describeTopics(final Collection<String> topicNames, DescribeTopicsOptions options); @Override DescribeClusterResult describeCluster(DescribeClusterOptions options); @Override DescribeAclsResult describeAcls(final AclBindingFilter filter, DescribeAclsOptions options); @Override CreateAclsResult createAcls(Collection<AclBinding> acls, CreateAclsOptions options); @Override DeleteAclsResult deleteAcls(Collection<AclBindingFilter> filters, DeleteAclsOptions options); @Override DescribeConfigsResult describeConfigs(Collection<ConfigResource> configResources, final DescribeConfigsOptions options); @Override AlterConfigsResult alterConfigs(Map<ConfigResource, Config> configs, final AlterConfigsOptions options);  }