Kube-side changes to #replicas should result in NotReady status

This changes the Topic Operator's handling of Kube-side changes to
KafkaTopic.spec.replicas by ensuring the reconciliation results in a
NotReady status. If the replication factor is changed on the Kafka side
then the reconciliation updates KafkaTopic.spec.replicas and results in
Ready status.

Note that, because the TO doesn't watch the relevent ZK nodes, the
KafkaTopic.spec.replicas only gets updated during a periodic reconciliation.

An integration test is added for the revised semantics.

Because change in replicas has never been supported I have also removed
unused the code which notionally allowed for the TO to perform partition
reassignment, but in fact was never used.

Fix #1847, fix #691

Signed-off-by: Tom Bentley <tbentley@redhat.com>

test/src/main/java/io/strimzi/test/k8s/OpenShift.java

@@ -25,7 +25,7 @@ public boolean isAvailable() {
     @Override
     public boolean isClusterUp() {
         try {
-            Exec.exec(OC, "status");
+            Exec.exec(OC, "cluster", "status");
             return true;
         } catch (KubeClusterException e) {
             if (e.result.exitStatus() == 1) {

topic-operator/pom.xml

@@ -115,10 +115,6 @@
             <groupId>io.strimzi</groupId>
             <artifactId>operator-common</artifactId>
         </dependency>
-        <dependency>
-            <groupId>com.github.spotbugs</groupId>
-            <artifactId>spotbugs-annotations</artifactId>
-        </dependency>
     </dependencies>
     <build>
         <plugins>

topic-operator/src/main/java/io/strimzi/operator/topic/K8sImpl.java

@@ -67,9 +67,9 @@ public Future<KafkaTopic> updateResource(KafkaTopic topicResource) {
             try {
                 KafkaTopic kafkaTopic = operation().inNamespace(namespace).withName(topicResource.getMetadata().getName()).patch(topicResource);
                 LOGGER.debug("KafkaTopic {} updated with version {}->{}",
-                        kafkaTopic.getMetadata().getName(),
+                        kafkaTopic != null && kafkaTopic.getMetadata() != null ? kafkaTopic.getMetadata().getName() : null,
                         topicResource.getMetadata() != null ? topicResource.getMetadata().getResourceVersion() : null,
-                        kafkaTopic.getMetadata().getResourceVersion());
+                        kafkaTopic != null && kafkaTopic.getMetadata() != null ? kafkaTopic.getMetadata().getResourceVersion() : null);
                 future.complete(kafkaTopic);
             } catch (Exception e) {
                 future.fail(e);

topic-operator/src/main/java/io/strimzi/operator/topic/Kafka.java

@@ -54,16 +54,6 @@ public interface Kafka {
      */
     Future<Void> increasePartitions(Topic topic);
 
-    /**
-     * Asynchronously change the topic's replication factor in Kafka,
-     * completing the returned Future when the topic has been updated.
-     * If the operation fails the returned Future will be failed with the
-     * KafkaException (not an ExecutionException).
-     * @param topic The topic.
-     * @return A future which is completed once the topic has been updated.
-     */
-    Future<Void> changeReplicationFactor(Topic topic);
-
     /**
      * Asynchronously fetch the topic metadata in Kafka,
      * completing the returned Future with the requested metadata.

topic-operator/src/main/java/io/strimzi/operator/topic/BaseKafkaImpl.java → topic-operator/src/main/java/io/strimzi/operator/topic/KafkaImpl.java

@@ -14,6 +14,8 @@
 import org.apache.kafka.clients.admin.Config;
 import org.apache.kafka.clients.admin.ListTopicsOptions;
 import org.apache.kafka.clients.admin.ListTopicsResult;
+import org.apache.kafka.clients.admin.NewPartitions;
+import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.clients.admin.TopicDescription;
 import org.apache.kafka.common.KafkaFuture;
 import org.apache.kafka.common.config.ConfigResource;
@@ -33,17 +35,17 @@
  * For example it is anticipated that one subclass will delegate to a "cluster balancer" so that cluster-wide,
  * traffic-aware assignments can be done.
  */
-public abstract class BaseKafkaImpl implements Kafka {
+public class KafkaImpl implements Kafka {
 
-    private final static Logger LOGGER = LogManager.getLogger(BaseKafkaImpl.class);
+    private final static Logger LOGGER = LogManager.getLogger(KafkaImpl.class);
 
     protected final AdminClient adminClient;
 
     protected final Vertx vertx;
 
     private volatile boolean stopped = false;
 
-    public BaseKafkaImpl(AdminClient adminClient, Vertx vertx) {
+    public KafkaImpl(AdminClient adminClient, Vertx vertx) {
         this.adminClient = adminClient;
         this.vertx = vertx;
     }
@@ -281,4 +283,30 @@ public Future<Set<String>> listTopics() {
     }
 
 
+    @Override
+    public Future<Void> increasePartitions(Topic topic) {
+        Future<Void> handler = Future.future();
+        final NewPartitions newPartitions = NewPartitions.increaseTo(topic.getNumPartitions());
+        final Map<String, NewPartitions> request = Collections.singletonMap(topic.getTopicName().toString(), newPartitions);
+        KafkaFuture<Void> future = adminClient.createPartitions(request).values().get(topic.getTopicName().toString());
+        queueWork(new UniWork<>("increasePartitions", future, handler));
+        return handler;
+    }
+
+    /**
+     * Create a new topic via the Kafka AdminClient API, calling the given handler
+     * (in a different thread) with the result.
+     */
+    @Override
+    public Future<Void> createTopic(Topic topic) {
+        Future<Void> handler = Future.future();
+        NewTopic newTopic = TopicSerialization.toNewTopic(topic, null);
+
+        LOGGER.debug("Creating topic {}", newTopic);
+        KafkaFuture<Void> future = adminClient.createTopics(
+                Collections.singleton(newTopic)).values().get(newTopic.name());
+        queueWork(new UniWork<>("createTopic", future, handler));
+        return handler;
+    }
+
 }