Fix Error State Issue when Kafka Cluster is Offline

kafka-transport/pom.xml

@@ -17,13 +17,6 @@
 			<version>${project.version}</version>
 			<scope>provided</scope>
 		</dependency>
-		<dependency>
-			<groupId>log4j</groupId>
-			<artifactId>log4j</artifactId>
-			<version>[1.2.9,)</version>
-			<scope>provided</scope>
-			<optional>true</optional>
-		</dependency>
 	</dependencies>
 	<build>
 		<plugins>

kafka-transport/src/main/java/com/esri/geoevent/transport/kafka/GEKafkaAdminUtil.java

@@ -0,0 +1,64 @@
+/*
+  Copyright 1995-2016 Esri
+    Licensed under the Apache License, Version 2.0 (the "License");
+    you may not use this file except in compliance with the License.
+    You may obtain a copy of the License at
+       http://www.apache.org/licenses/LICENSE-2.0
+    Unless required by applicable law or agreed to in writing, software
+    distributed under the License is distributed on an "AS IS" BASIS,
+    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+    See the License for the specific language governing permissions and
+    limitations under the License.
+  For additional information, contact:
+  Environmental Systems Research Institute, Inc.
+  Attn: Contracts Dept
+  380 New York Street
+  Redlands, California, USA 92373
+  email: [email protected]
+*/
+
+package com.esri.geoevent.transport.kafka;
+
+import org.apache.kafka.clients.ApiVersions;
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.DescribeClusterOptions;
+import org.apache.kafka.clients.admin.DescribeClusterResult;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.protocol.ApiKeys;
+
+import java.util.*;
+
+public class GEKafkaAdminUtil
+{
+
+  public static void performAdminClientValidation(Properties configProperties)
+  {
+    AdminClient adminClient = AdminClient.create(configProperties);
+    DescribeClusterOptions describeOptions = new DescribeClusterOptions();
+    describeOptions.timeoutMs(Integer.valueOf(10));
+    Timer timer = new Timer("ReconnectionTimer");
+    GEKafkaTimerTask timerTask = new GEKafkaTimerTask(adminClient, timer);
+    timer.scheduleAtFixedRate(timerTask, 0L, 10000L);
+  }
+
+  public static boolean checkAPILevelCompatibility(Node... kafkaNodes)
+  {
+    ApiVersions apiVersions = new ApiVersions();
+    ArrayList<Short> apiVersionCollection = new ArrayList<>();
+    ApiKeys apiKeys = ApiKeys.forId(0);
+    Arrays.stream(kafkaNodes).forEach(eachNode -> apiVersionCollection.add(apiVersions.get(eachNode.idString()).latestUsableVersion(apiKeys)));
+    long count = apiVersionCollection.stream().filter(predicate -> predicate.intValue() > ((short) 0.11)).count();
+    return count > 0L ? true : false;
+  }
+
+  public static synchronized Collection<Node> listAllNodesInCluster(Properties configProperties)
+  {
+    AdminClient adminClient = AdminClient.create(configProperties);
+    DescribeClusterOptions describeOptions = new DescribeClusterOptions();
+    describeOptions.timeoutMs(Integer.valueOf(10));
+    DescribeClusterResult describedCluster = adminClient.describeCluster(describeOptions);
+    ArrayList<Node> nodeArrayList = new ArrayList<>();
+    describedCluster.nodes().whenComplete((nodes, throwable) -> nodeArrayList.addAll(nodes));
+    return nodeArrayList;
+  }
+}

kafka-transport/src/main/java/com/esri/geoevent/transport/kafka/GEKafkaTimerTask.java

@@ -0,0 +1,70 @@
+/*
+  Copyright 1995-2016 Esri
+    Licensed under the Apache License, Version 2.0 (the "License");
+    you may not use this file except in compliance with the License.
+    You may obtain a copy of the License at
+       http://www.apache.org/licenses/LICENSE-2.0
+    Unless required by applicable law or agreed to in writing, software
+    distributed under the License is distributed on an "AS IS" BASIS,
+    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+    See the License for the specific language governing permissions and
+    limitations under the License.
+  For additional information, contact:
+  Environmental Systems Research Institute, Inc.
+  Attn: Contracts Dept
+  380 New York Street
+  Redlands, California, USA 92373
+  email: [email protected]
+*/
+
+
+package com.esri.geoevent.transport.kafka;
+
+import com.esri.ges.framework.i18n.BundleLogger;
+import com.esri.ges.framework.i18n.BundleLoggerFactory;
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.DescribeClusterResult;
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.Node;
+
+import java.util.Collection;
+import java.util.Timer;
+import java.util.TimerTask;
+import java.util.concurrent.ExecutionException;
+
+public class GEKafkaTimerTask extends TimerTask
+{
+
+  private static final BundleLogger LOGGER          = BundleLoggerFactory.getLogger(GEKafkaTimerTask.class);
+  private              AdminClient  adminClient;
+  private              Timer        timer;
+
+  public GEKafkaTimerTask(AdminClient adminClient, Timer timer)
+  {
+    this.adminClient = adminClient;
+    this.timer = timer;
+  }
+
+  @Override
+  public void run()
+  {
+    DescribeClusterResult clusterDescription = adminClient.describeCluster();
+    KafkaFuture<Collection<Node>> clusterNodes = clusterDescription.nodes();
+    try
+    {
+
+      long count = clusterNodes.get().stream().count();
+      if (count > 0)
+      {
+        cancel();
+        timer.cancel();
+        timer.purge();
+      }
+    }
+    catch (InterruptedException | ExecutionException intEx)
+    {
+      LOGGER.error("PROBLEM_WITH_CLUSTER", intEx);
+    }
+
+  }
+}

kafka-transport/src/main/java/com/esri/geoevent/transport/kafka/GeoEventKafkaConsumerRebalancer.java

@@ -0,0 +1,59 @@
+/*
+  Copyright 1995-2016 Esri
+    Licensed under the Apache License, Version 2.0 (the "License");
+    you may not use this file except in compliance with the License.
+    You may obtain a copy of the License at
+       http://www.apache.org/licenses/LICENSE-2.0
+    Unless required by applicable law or agreed to in writing, software
+    distributed under the License is distributed on an "AS IS" BASIS,
+    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+    See the License for the specific language governing permissions and
+    limitations under the License.
+  For additional information, contact:
+  Environmental Systems Research Institute, Inc.
+  Attn: Contracts Dept
+  380 New York Street
+  Redlands, California, USA 92373
+  email: [email protected]
+*/
+
+package com.esri.geoevent.transport.kafka;
+
+import com.esri.ges.framework.i18n.BundleLogger;
+import com.esri.ges.framework.i18n.BundleLoggerFactory;
+import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.TopicPartition;
+
+import java.util.Collection;
+
+public class GeoEventKafkaConsumerRebalancer implements ConsumerRebalanceListener
+{
+
+  private boolean                       isFromBeginning;
+  private KafkaConsumer<byte[], byte[]> consumer;
+  private BundleLogger                  LOGGER = BundleLoggerFactory.getLogger(GeoEventKafkaConsumerRebalancer.class);
+
+
+  public GeoEventKafkaConsumerRebalancer(boolean fromBeginning, KafkaConsumer<byte[], byte[]> kafkaConsumer)
+  {
+    isFromBeginning = fromBeginning;
+    consumer = kafkaConsumer;
+  }
+
+
+  @Override
+  public void onPartitionsRevoked(Collection<TopicPartition> partitions)
+  {
+    partitions.forEach(topicPartition -> LOGGER.info("PARTITION_REVOKED_ON_TOPIC: " + topicPartition.topic()));
+  }
+
+  @Override
+  public void onPartitionsAssigned(Collection<TopicPartition> partitions)
+  {
+    if (isFromBeginning)
+      consumer.seekToBeginning(partitions);
+    partitions.forEach(topicPartition -> LOGGER.info("PARTITION_ASSIGNED_ON_TOPIC: " + topicPartition.topic()));
+  }
+
+}