FIX: unsupported message format in kafka

Signed-off-by: George Chen <[email protected]>

data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/util/MessageFormat.java

@@ -9,6 +9,7 @@
 
 import java.util.Arrays;
 import java.util.Map;
+import java.util.Optional;
 import java.util.stream.Collectors;
 
 /**
@@ -33,6 +34,7 @@ public enum MessageFormat {
 
     @JsonCreator
     public static MessageFormat getByMessageFormatByName(final String name) {
-        return MESSAGE_FORMAT_MAP.get(name.toLowerCase());
+        return Optional.ofNullable(MESSAGE_FORMAT_MAP.get(name.toLowerCase())).orElseThrow(
+                () -> new IllegalArgumentException("Unsupported message format in kafka plugin: " + name));
     }
 }

data-prepper-plugins/kafka-plugins/src/test/java/org/opensearch/dataprepper/plugins/kafka/util/MessageFormatTest.java

@@ -5,19 +5,26 @@
 
 package org.opensearch.dataprepper.plugins.kafka.util;
 
+import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.EnumSource;
 import org.mockito.junit.jupiter.MockitoExtension;
 
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 
 @ExtendWith(MockitoExtension.class)
 class MessageFormatTest {
 	@ParameterizedTest
 	@EnumSource(MessageFormat.class)
-	void getByNameTest(final MessageFormat name) {
+	void getByNameSupportedTest(final MessageFormat name) {
 		assertThat(MessageFormat.getByMessageFormatByName(name.name()), is(name));
 	}
+
+	@Test
+	void getByNameUnsupportedTest() {
+		assertThrows(IllegalArgumentException.class, () -> MessageFormat.getByMessageFormatByName("unknown"));
+	}
 }