Merge pull request #1382 from zalando/nakadi-batch-published-to-avro

Bugfix: metadata should have version of event-type schema
zalando · Mar 30, 2022 · d95baf9 · d95baf9
2 parents ed827c3 + 0a316d6
commit d95baf9
Showing 1 changed file with 8 additions and 6 deletions.
diff --git a/core-services/src/main/java/org/zalando/nakadi/service/publishing/NakadiKpiPublisher.java b/core-services/src/main/java/org/zalando/nakadi/service/publishing/NakadiKpiPublisher.java
@@ -87,11 +87,12 @@ public void publishNakadiBatchPublishedEvent(
                     avroSchema.getLatestEventTypeSchemaVersion(AvroSchema.METADATA_KEY);
             final byte metadataVersion = Byte.parseByte(latestMeta.getKey());
 
-            final GenericRecord metadata = buildMetaDataGenericRecord(
-                    KPIEventTypes.BATCH_PUBLISHED, latestMeta.getValue(), latestMeta.getKey());
-
             final Map.Entry<String, Schema> latestSchema =
                     avroSchema.getLatestEventTypeSchemaVersion(KPIEventTypes.BATCH_PUBLISHED);
+
+            final GenericRecord metadata = buildMetaDataGenericRecord(
+                    KPIEventTypes.BATCH_PUBLISHED, latestMeta.getValue(), latestSchema.getKey());
+
             final GenericRecord event = new GenericRecordBuilder(latestSchema.getValue())
                     .set("event_type", eventTypeName)
                     .set("app", applicationName)
@@ -143,12 +144,13 @@ public void publishAccessLogEvent(final String method,
                     avroSchema.getLatestEventTypeSchemaVersion(AvroSchema.METADATA_KEY);
             final byte metadataVersion = Byte.parseByte(latestMeta.getKey());
 
+            final Map.Entry<String, Schema> latestSchema =
+                    avroSchema.getLatestEventTypeSchemaVersion(KPIEventTypes.ACCESS_LOG);
+
             final GenericRecord metadata = buildMetaDataGenericRecord(
-                    KPIEventTypes.ACCESS_LOG, latestMeta.getValue(), latestMeta.getKey(), user);
+                    KPIEventTypes.ACCESS_LOG, latestMeta.getValue(), latestSchema.getKey(), user);
 
 
-            final Map.Entry<String, Schema> latestSchema =
-                    avroSchema.getLatestEventTypeSchemaVersion(KPIEventTypes.ACCESS_LOG);
             final GenericRecord event = new GenericRecordBuilder(latestSchema.getValue())
                     .set("method", method)
                     .set("path", path)