Fix strimzi#1157

Signed-off-by: Tom Bentley <tbentley@redhat.com>
tombentley · Nov 19, 2020 · a0f2a2f · a0f2a2f
1 parent 6db173e
commit a0f2a2f
Show file tree

Hide file tree

Showing 4 changed files with 45 additions and 22 deletions.
diff --git a/topic-operator/src/main/java/io/strimzi/operator/topic/Kafka.java b/topic-operator/src/main/java/io/strimzi/operator/topic/Kafka.java
@@ -34,6 +34,16 @@ public interface Kafka {
      */
     Future<Void> deleteTopic(TopicName topicName);
 
+    /**
+     * Wait for the given topic to not existing Kafka ,
+     * completing the returned Future when the topic does not exists.
+     * If the operation fails the returned Future will be failed with the
+     * KafkaException (not an ExecutionException).
+     * @param topicName The name of the topic to delete.
+     * @return A future which is completed once the topic has been deleted.
+     */
+    Future<Void> awaitNotExists(TopicName topicName);
+
     /**
      * Asynchronously update the topic config in Kafka,
      * completing the returned Future when the topic has been updated.

diff --git a/topic-operator/src/main/java/io/strimzi/operator/topic/KafkaImpl.java b/topic-operator/src/main/java/io/strimzi/operator/topic/KafkaImpl.java
@@ -58,21 +58,7 @@ public Future<Void> deleteTopic(TopicName topicName) {
         KafkaFuture<Void> future = adminClient.deleteTopics(
                 singleton(topicName.toString())).values().get(topicName.toString());
         mapFuture(future).compose(ig ->
-                Util.waitFor(vertx, "deleted sync " + topicName, "deleted", 1000, 120_000, () -> {
-                    try {
-                        return adminClient.describeTopics(singleton(topicName.toString())).all().get().get(topicName.toString()) == null;
-                    } catch (ExecutionException e) {
-                        if (e.getCause() instanceof UnknownTopicOrPartitionException) {
-                            return true;
-                        } else if (e.getCause() instanceof RuntimeException) {
-                            throw (RuntimeException) e.getCause();
-                        } else {
-                            throw new RuntimeException(e);
-                        }
-                    } catch (InterruptedException e) {
-                        throw new RuntimeException(e);
-                    }
-                }, error -> true)
+                awaitNotExists(topicName)
         ).onComplete(ar -> {
             // Complete the result future on the context thread.
             vertx.runOnContext(ignored -> {
@@ -82,6 +68,24 @@ public Future<Void> deleteTopic(TopicName topicName) {
         return handler.future();
     }
 
+    public Future<Void> awaitNotExists(TopicName topicName) {
+        return Util.waitFor(vertx, "deleted sync " + topicName, "deleted", 1000, 300_000, () -> {
+            try {
+                return adminClient.describeTopics(singleton(topicName.toString())).all().get().get(topicName.toString()) == null;
+            } catch (ExecutionException e) {
+                if (e.getCause() instanceof UnknownTopicOrPartitionException) {
+                    return true;
+                } else if (e.getCause() instanceof RuntimeException) {
+                    throw (RuntimeException) e.getCause();
+                } else {
+                    throw new RuntimeException(e);
+                }
+            } catch (InterruptedException e) {
+                throw new RuntimeException(e);
+            }
+        }, error -> true);
+    }
+
     @SuppressWarnings("deprecation")
     @Override
     public Future<Void> updateTopicConfig(Topic topic) {

diff --git a/topic-operator/src/main/java/io/strimzi/operator/topic/TopicOperator.java b/topic-operator/src/main/java/io/strimzi/operator/topic/TopicOperator.java
@@ -771,13 +771,17 @@ void enqueue(Handler<Void> event) {
 
     /** Called when a topic znode is deleted in ZK */
     Future<Void> onTopicDeleted(LogContext logContext, TopicName topicName) {
-        return executeWithTopicLockHeld(logContext, topicName,
-            new Reconciliation("onTopicDeleted", true) {
-                @Override
-                public Future<Void> execute() {
-                    return reconcileOnTopicChange(logContext, topicName, null, this);
-                }
-            });
+        return kafka.awaitNotExists(topicName).compose(
+            ignored ->
+                executeWithTopicLockHeld(logContext, topicName,
+                    new Reconciliation("onTopicDeleted", true) {
+                        @Override
+                        public Future<Void> execute() {
+                            return reconcileOnTopicChange(logContext, topicName, null, this);
+                        }
+                    }),
+            error ->
+                Future.failedFuture("Ignored spurious-seeming topic deletion"));
     }
 
     private Map<String, Long> statusUpdateGeneration = new HashMap<>();

diff --git a/topic-operator/src/test/java/io/strimzi/operator/topic/MockKafka.java b/topic-operator/src/test/java/io/strimzi/operator/topic/MockKafka.java
@@ -153,6 +153,11 @@ public Future<Void> deleteTopic(TopicName topicName) {
         return event;
     }
 
+    @Override
+    public Future<Void> awaitNotExists(TopicName topicName) {
+        return Future.succeededFuture();
+    }
+
     public MockKafka setUpdateTopicResponse(Function<TopicName, Future<Void>> updateTopicResponse) {
         this.updateTopicResponse = updateTopicResponse;
         return this;