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

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;
+    }
+
 }