kowshik commented on a change in pull request #9001:
URL: https://github.com/apache/kafka/pull/9001#discussion_r463916219
##########
File path:
clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
##########
@@ -3615,6 +3662,137 @@ public void testListOffsetsNonRetriableErrors() throws
Exception {
}
}
+ @Test
+ public void testUpdateFeaturesDuringSuccess() throws Exception {
+ testUpdateFeatures(
+ makeTestFeatureUpdates(),
+ makeTestFeatureUpdateErrors(Errors.NONE));
+ }
+
+ @Test
+ public void testUpdateFeaturesInvalidRequestError() throws Exception {
+ testUpdateFeatures(
+ makeTestFeatureUpdates(),
+ makeTestFeatureUpdateErrors(Errors.INVALID_REQUEST));
+ }
+
+ @Test
+ public void testUpdateFeaturesUpdateFailedError() throws Exception {
+ testUpdateFeatures(
+ makeTestFeatureUpdates(),
+ makeTestFeatureUpdateErrors(Errors.FEATURE_UPDATE_FAILED));
+ }
+
+ @Test
+ public void testUpdateFeaturesPartialSuccess() throws Exception {
+ final Map<String, Errors> errors =
makeTestFeatureUpdateErrors(Errors.NONE);
+ errors.put("test_feature_2", Errors.INVALID_REQUEST);
+ testUpdateFeatures(makeTestFeatureUpdates(), errors);
+ }
+
+ private Map<String, FeatureUpdate> makeTestFeatureUpdates() {
+ return Utils.mkMap(
+ Utils.mkEntry("test_feature_1", new FeatureUpdate((short) 2,
false)),
+ Utils.mkEntry("test_feature_2", new FeatureUpdate((short) 3,
true)));
+ }
+
+ private Map<String, Errors> makeTestFeatureUpdateErrors(final Errors
error) {
+ final Map<String, FeatureUpdate> updates = makeTestFeatureUpdates();
+ final Map<String, Errors> errors = new HashMap<>();
+ for (Map.Entry<String, FeatureUpdate> entry : updates.entrySet()) {
+ errors.put(entry.getKey(), error);
+ }
+ return errors;
+ }
+
+ private void testUpdateFeatures(Map<String, FeatureUpdate> featureUpdates,
+ Map<String, Errors> featureUpdateErrors)
throws Exception {
+ try (final AdminClientUnitTestEnv env = mockClientEnv()) {
+ env.kafkaClient().prepareResponse(
+ body -> body instanceof UpdateFeaturesRequest,
+ prepareUpdateFeaturesResponse(featureUpdateErrors));
+ final Map<String, KafkaFuture<Void>> futures =
env.adminClient().updateFeatures(
+ featureUpdates,
+ new UpdateFeaturesOptions().timeoutMs(10000)).values();
+ for (Map.Entry<String, KafkaFuture<Void>> entry :
futures.entrySet()) {
+ final KafkaFuture<Void> future = entry.getValue();
+ final Errors error = featureUpdateErrors.get(entry.getKey());
+ if (error == Errors.NONE) {
+ future.get();
+ } else {
+ final ExecutionException e =
assertThrows(ExecutionException.class,
+ () -> future.get());
Review comment:
Isn't that what I'm using currently?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
