KAFKA-14712: Produce correct error msg with correct metadataversion

metadata/src/main/java/org/apache/kafka/image/writer/ImageWriterOptions.java

@@ -29,6 +29,7 @@
 public final class ImageWriterOptions {
     public static class Builder {
         private MetadataVersion metadataVersion;
+        private MetadataVersion requestedMetadataVersion;
         private Consumer<UnwritableMetadataException> lossHandler = e -> {
             throw e;
         };
@@ -42,6 +43,7 @@ public Builder(MetadataImage image) {
         }
 
         public Builder setMetadataVersion(MetadataVersion metadataVersion) {
+            setRequestedMetadataVersion(metadataVersion);
             if (metadataVersion.isLessThan(MetadataVersion.MINIMUM_BOOTSTRAP_VERSION)) {
                 // When writing an image, all versions less than 3.3-IV0 are treated as 3.0-IV1.
                 // This is because those versions don't support FeatureLevelRecord.
@@ -58,37 +60,48 @@ public Builder setRawMetadataVersion(MetadataVersion metadataVersion) {
             return this;
         }
 
+        public void setRequestedMetadataVersion(MetadataVersion orgMetadataVersion) {
+            this.requestedMetadataVersion = orgMetadataVersion;
+        }
+
         public MetadataVersion metadataVersion() {
             return metadataVersion;
         }
 
+        public MetadataVersion requestedMetadataVersion() {
+            return requestedMetadataVersion;
+        }
+
         public Builder setLossHandler(Consumer<UnwritableMetadataException> lossHandler) {
             this.lossHandler = lossHandler;
             return this;
         }
 
         public ImageWriterOptions build() {
-            return new ImageWriterOptions(metadataVersion, lossHandler);
+            return new ImageWriterOptions(metadataVersion, lossHandler, requestedMetadataVersion);
         }
     }
 
     private final MetadataVersion metadataVersion;
+    private final MetadataVersion requestedMetadataVersion;
     private final Consumer<UnwritableMetadataException> lossHandler;
 
     private ImageWriterOptions(
         MetadataVersion metadataVersion,
-        Consumer<UnwritableMetadataException> lossHandler
+        Consumer<UnwritableMetadataException> lossHandler,
+        MetadataVersion orgMetadataVersion
     ) {
         this.metadataVersion = metadataVersion;
         this.lossHandler = lossHandler;
+        this.requestedMetadataVersion = orgMetadataVersion;
     }
 
     public MetadataVersion metadataVersion() {
         return metadataVersion;
     }
 
     public void handleLoss(String loss) {
-        lossHandler.accept(new UnwritableMetadataException(metadataVersion, loss));
+        lossHandler.accept(new UnwritableMetadataException(requestedMetadataVersion, loss));
     }
 }
 

metadata/src/test/java/org/apache/kafka/image/ImageDowngradeTest.java

@@ -138,7 +138,7 @@ private static void writeWithExpectedLosses(
         MetadataImage image = delta.apply(MetadataProvenance.EMPTY);
         RecordListWriter writer = new RecordListWriter();
         image.write(writer, new ImageWriterOptions.Builder().
-                setRawMetadataVersion(metadataVersion).
+                setMetadataVersion(metadataVersion).
                 setLossHandler(lossConsumer).
                 build());
         assertEquals(expectedLosses, lossConsumer.losses, "Failed to get expected metadata losses.");

metadata/src/test/java/org/apache/kafka/image/writer/ImageWriterOptionsTest.java

@@ -21,6 +21,8 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.Timeout;
 
+import java.util.function.Consumer;
+
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
@@ -44,9 +46,28 @@ public void testSetMetadataVersion() {
                     setMetadataVersion(version);
             if (i < MetadataVersion.MINIMUM_BOOTSTRAP_VERSION.ordinal()) {
                 assertEquals(MetadataVersion.MINIMUM_KRAFT_VERSION, options.metadataVersion());
+                assertEquals(version, options.requestedMetadataVersion());
             } else {
                 assertEquals(version, options.metadataVersion());
             }
         }
     }
+
+    @Test
+    public void testHandleLoss() {
+        String expectedMessage = "stuff";
+
+        for (int i = MetadataVersion.MINIMUM_KRAFT_VERSION.ordinal();
+             i < MetadataVersion.VERSIONS.length;
+             i++) {
+            MetadataVersion version = MetadataVersion.VERSIONS[i];
+            String formattedMessage = String.format("Metadata has been lost because the following could not be represented in metadata version %s: %s", version, expectedMessage);
+            Consumer<UnwritableMetadataException> customLossHandler = e -> assertEquals(formattedMessage, e.getMessage());
+            ImageWriterOptions options = new ImageWriterOptions.Builder()
+                    .setMetadataVersion(version)
+                    .setLossHandler(customLossHandler)
+                    .build();
+            options.handleLoss(expectedMessage);
+        }
+    }
 }