Testing that every schema version is loadable

core-common/src/main/java/org/zalando/nakadi/config/KPIEventTypes.java

@@ -3,4 +3,7 @@
 public final class KPIEventTypes {
     public static final String ACCESS_LOG = "nakadi.access.log";
     public static final String BATCH_PUBLISHED = "nakadi.batch.published";
+    public static final String DATA_STREAMED = "nakadi.data.streamed";
+    public static final String EVENT_TYPE_LOG = "nakadi.event.type.log";
+    public static final String SUBSCRIPTION_LOG = "nakadi.subscription.log";
 }

core-common/src/main/java/org/zalando/nakadi/service/AvroSchema.java

@@ -31,7 +31,10 @@ public class AvroSchema {
     private static final Collection<String> INTERNAL_EVENT_TYPE_NAMES = Set.of(
             METADATA_KEY,
             KPIEventTypes.ACCESS_LOG,
-            KPIEventTypes.BATCH_PUBLISHED);
+            KPIEventTypes.BATCH_PUBLISHED,
+            KPIEventTypes.DATA_STREAMED,
+            KPIEventTypes.EVENT_TYPE_LOG,
+            KPIEventTypes.SUBSCRIPTION_LOG);
 
     private final Map<String, TreeMap<String, Schema>> eventTypeSchema;
     private final AvroMapper avroMapper;

core-common/src/main/resources/event-type-schema/nakadi.data.streamed/nakadi.data.streamed.0.avsc

@@ -48,4 +48,4 @@
       "type": "int"
     }
   ]
-}
+}

core-common/src/main/resources/event-type-schema/nakadi.event.type.log/nakadi.event.type.log.0.avsc

@@ -24,4 +24,4 @@
       "type": "string"
     }
   ]
-}
+}

core-common/src/main/resources/event-type-schema/nakadi.subscription.log/nakadi.subscription.log.0.avsc

@@ -12,4 +12,4 @@
       "type": "string"
     }
   ]
-}
+}

core-common/src/test/java/org/zalando/nakadi/service/AvroSchemaTest.java

@@ -0,0 +1,40 @@
+package org.zalando.nakadi.service;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.dataformat.avro.AvroMapper;
+import org.junit.Test;
+import org.junit.jupiter.api.Assertions;
+import org.springframework.core.io.DefaultResourceLoader;
+import org.zalando.nakadi.config.KPIEventTypes;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+public class AvroSchemaTest {
+    private final AvroSchema avroSchema;
+    private final Map<String, List<String>> allSchemas = Map.of(
+            AvroSchema.METADATA_KEY, List.of("0", "1"),
+            KPIEventTypes.ACCESS_LOG, List.of("0", "1"),
+            KPIEventTypes.BATCH_PUBLISHED, List.of("0"),
+            KPIEventTypes.DATA_STREAMED, List.of("0"),
+            KPIEventTypes.EVENT_TYPE_LOG, List.of("0"),
+            KPIEventTypes.SUBSCRIPTION_LOG, List.of("0")
+    );
+
+    public AvroSchemaTest() throws IOException {
+        final var eventTypeRes = new DefaultResourceLoader().getResource("event-type-schema/");
+        avroSchema = new AvroSchema(new AvroMapper(), new ObjectMapper(), eventTypeRes);
+    }
+
+    @Test
+    public void testAllSchemaVersionAreLoadable() {
+        allSchemas.entrySet().stream()
+                .flatMap(schemaEntry -> schemaEntry.getValue().stream()
+                        .map(version -> Map.entry(schemaEntry.getKey(), version)))
+                .forEach(schemaEntry -> Assertions.assertDoesNotThrow(
+                        () -> avroSchema.getEventTypeSchema(schemaEntry.getKey(), schemaEntry.getValue()),
+                        "Schema of " + schemaEntry.getKey()
+                                + ", version " + schemaEntry.getValue() + " unavailable"));
+    }
+}

core-common/src/test/resources/log4j.properties

@@ -0,0 +1,25 @@
+log4j.rootCategory=INFO, CONSOLE
+
+LOG_PATTERN=[%d{yyyy-MM-dd HH:mm:ss.SSSXXX}] [%p] [%x] [%t] [%c] --- %m %throwable{compact} %n
+
+# CONSOLE is set to be a ConsoleAppender using a PatternLayout.
+log4j.appender.CONSOLE=org.apache.log4j.ConsoleAppender
+log4j.appender.CONSOLE.layout=org.apache.log4j.EnhancedPatternLayout
+log4j.appender.CONSOLE.layout.ConversionPattern=${LOG_PATTERN}
+
+log4j.category.org.apache.catalina.startup.DigesterFactory=ERROR
+log4j.category.org.apache.catalina.util.LifecycleBase=ERROR
+log4j.category.org.apache.coyote.http11.Http11NioProtocol=WARN
+log4j.category.org.apache.tomcat.util.net.NioSelectorPool=WARN
+log4j.category.org.crsh.plugin=WARN
+log4j.category.org.crsh.ssh=WARN
+log4j.category.org.eclipse.jetty.util.component.AbstractLifeCycle=ERROR
+log4j.category.org.hibernate.validator.internal.util.Version=WARN
+log4j.category.org.springframework.boot.actuate.autoconfigure.CrshAutoConfiguration=WARN
+log4j.category.org.springframework.boot.actuate.endpoint.jmx=WARN
+log4j.category.org.thymeleaf=WARN
+log4j.category.org.zalando.nakadi=DEBUG
+log4j.category.org.zalando.nakadi.config=INFO
+log4j.category.org.apache.kafka=WARN
+log4j.category.org.zalando.nakadi.service.ClosedConnectionsCrutch=WARN
+log4j.category.org.zalando.nakadi.repository.kafka.KafkaFactory$KafkaCrutchConsumer=WARN