[FSTORE-1011] Validation message during data ingestion to FG when the…

…re are no kafka topics configured (#1414)
logicalclocks · Oct 16, 2023 · d06ddfe · d06ddfe
1 parent bfb3da3
commit d06ddfe
Show file tree

Hide file tree

Showing 3 changed files with 20 additions and 3 deletions.
diff --git a/hopsworks-common/src/main/java/io/hops/hopsworks/common/kafka/KafkaController.java b/hopsworks-common/src/main/java/io/hops/hopsworks/common/kafka/KafkaController.java
@@ -132,6 +132,9 @@ public synchronized ProjectTopics createTopic(Project project, TopicDTO topicDto
 
   protected void checkReplication(TopicDTO topicDto) throws KafkaException {
     List<String> brokerEndpoints = kafkaBrokers.getBrokerEndpoints(KafkaBrokers.BrokerProtocol.INTERNAL);
+    if (brokerEndpoints.isEmpty()) {
+      throw new KafkaException(RESTCodes.KafkaErrorCode.BROKER_MISSING, Level.FINE);
+    }
     if (brokerEndpoints.size() < topicDto.getNumOfReplicas()) {
       throw new KafkaException(RESTCodes.KafkaErrorCode.TOPIC_REPLICATION_ERROR, Level.FINE,
           "maximum: " + brokerEndpoints.size());

diff --git a/hopsworks-common/src/test/io/hops/hopsworks/common/kafka/TestKafkaController.java b/hopsworks-common/src/test/io/hops/hopsworks/common/kafka/TestKafkaController.java
@@ -48,7 +48,6 @@ public void setup() {
   @Test
   public void testCheckReplication() throws KafkaException {
     // Arrange
-    Mockito.doReturn(false).when(kafkaController).externalKafka(Mockito.any());
     List<String> brokers = Collections.nCopies(10, "endpoint");
     Mockito.doReturn(brokers).when(kafkaController.kafkaBrokers).getBrokerEndpoints(Mockito.any());
     TopicDTO topicDTO = new TopicDTO();
@@ -64,7 +63,6 @@ public void testCheckReplication() throws KafkaException {
   @Test(expected = KafkaException.class)
   public void testCheckReplicationTooFewBrokers() throws KafkaException {
     // Arrange
-    Mockito.doReturn(false).when(kafkaController).externalKafka(Mockito.any());
     List<String> brokers = Collections.nCopies(1, "endpoint");
     Mockito.doReturn(brokers).when(kafkaController.kafkaBrokers).getBrokerEndpoints(Mockito.any());
     TopicDTO topicDTO = new TopicDTO();
@@ -76,6 +74,19 @@ public void testCheckReplicationTooFewBrokers() throws KafkaException {
     // Assert
   }
 
+  @Test(expected = KafkaException.class)
+  public void testCheckReplicationNoBrokers() throws KafkaException {
+    // Arrange
+    Mockito.doReturn(new ArrayList<>()).when(kafkaController.kafkaBrokers).getBrokerEndpoints(Mockito.any());
+    TopicDTO topicDTO = new TopicDTO();
+    topicDTO.setNumOfReplicas(10);
+
+    // Act
+    kafkaController.checkReplication(topicDTO);
+
+    // Assert
+  }
+
   @Test
   public void testRemoveKafkaTopicsInternalKafka() {
     // Arrange
@@ -142,6 +153,8 @@ public void testRemoveTopicFromProjectExternalKafka() throws KafkaException {
   @Test(expected = NullPointerException.class)
   public void testCreateTopicInternalKafka() throws KafkaException {
     // Arrange
+    List<String> brokers = Collections.nCopies(10, "endpoint");
+    Mockito.doReturn(brokers).when(kafkaController.kafkaBrokers).getBrokerEndpoints(Mockito.any());
     Mockito.doReturn(false).when(kafkaController).externalKafka(Mockito.any());
 
     // Act

diff --git a/hopsworks-rest-utils/src/main/java/io/hops/hopsworks/restutils/RESTCodes.java b/hopsworks-rest-utils/src/main/java/io/hops/hopsworks/restutils/RESTCodes.java
@@ -760,7 +760,8 @@ public enum KafkaErrorCode implements RESTErrorCode {
       Response.Status.SERVICE_UNAVAILABLE),
     TOPIC_DELETION_FAILED(22, "Could not delete Kafka topics.", Response.Status.INTERNAL_SERVER_ERROR),
     TOPIC_FETCH_FAILED(23, "Could not fetch topic details.", Response.Status.INTERNAL_SERVER_ERROR),
-    TOPIC_CREATION_FAILED(24, "Could not create topic.", Response.Status.INTERNAL_SERVER_ERROR);
+    TOPIC_CREATION_FAILED(24, "Could not create topic.", Response.Status.INTERNAL_SERVER_ERROR),
+    BROKER_MISSING(25, "Could not find a broker endpoint.", Response.Status.NOT_FOUND);
 
 
     private Integer code;