diff --git a/CHANGES.md b/CHANGES.md
index 87b8ad672ebb..40bdd69c86d7 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -81,6 +81,7 @@
 ## Bugfixes
 
 * (Go) Fixed a data race on the Prism runner's artifact cache map in JobServices ([#32656](https://github.com/apache/beam/issues/32656)).
+* (Java) Fixed the declared schema of the error output of the Kafka write SchemaTransform, which wrapped the error schema a second time and did not match the rows it emits ([#39760](https://github.com/apache/beam/issues/39760)).
 * Fixed X (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
 
 ## Security Fixes
diff --git a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaWriteSchemaTransformProvider.java b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaWriteSchemaTransformProvider.java
index d8d0fe478e35..e59159ba5b81 100644
--- a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaWriteSchemaTransformProvider.java
+++ b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaWriteSchemaTransformProvider.java
@@ -294,8 +294,7 @@ public PCollectionRowTuple expand(PCollectionRowTuple input) {
       }
 
       // TODO: include output from KafkaIO Write once updated from PDone
-      PCollection<Row> errorOutput =
-          outputTuple.get(ERROR_TAG).setRowSchema(ErrorHandling.errorSchema(errorSchema));
+      PCollection<Row> errorOutput = outputTuple.get(ERROR_TAG).setRowSchema(errorSchema);
       return PCollectionRowTuple.of(
           handleErrors ? configuration.getErrorHandling().getOutput() : "errors", errorOutput);
     }
diff --git a/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaWriteSchemaTransformProviderTest.java b/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaWriteSchemaTransformProviderTest.java
index cc5bd82a5a88..ef53ff0bb83c 100644
--- a/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaWriteSchemaTransformProviderTest.java
+++ b/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaWriteSchemaTransformProviderTest.java
@@ -50,6 +50,7 @@
 import org.apache.beam.sdk.transforms.SerializableFunction;
 import org.apache.beam.sdk.values.KV;
 import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionRowTuple;
 import org.apache.beam.sdk.values.PCollectionTuple;
 import org.apache.beam.sdk.values.Row;
 import org.apache.beam.sdk.values.TupleTag;
@@ -280,6 +281,29 @@ public void testBuildTransformWithManaged() {
     }
   }
 
+  @Test
+  public void testErrorOutputCarriesTheSchemaErrorCounterFnEmits() {
+    // The output schema is fixed while the graph is built, so this needs no runner.
+    p.enableAbandonedNodeEnforcement(false);
+
+    Schema inputSchema = Schema.builder().addByteArrayField("bytes").build();
+    KafkaWriteSchemaTransformProvider.KafkaWriteSchemaTransformConfiguration configuration =
+        KafkaWriteSchemaTransformProvider.KafkaWriteSchemaTransformConfiguration.builder()
+            .setFormat("RAW")
+            .setTopic("test-topic")
+            .setBootstrapServers("host:9092")
+            .setErrorHandling(ErrorHandling.builder().setOutput("errors").build())
+            .build();
+
+    PCollectionRowTuple output =
+        PCollectionRowTuple.of("input", p.apply(Create.empty(inputSchema)))
+            .apply(new KafkaWriteSchemaTransformProvider().from(configuration));
+
+    // ErrorCounterFn emits ErrorHandling.errorRecord(errorSchema, ...), where errorSchema is
+    // already ErrorHandling.errorSchema(inputSchema).
+    assertEquals(ErrorHandling.errorSchema(inputSchema), output.get("errors").getSchema());
+  }
+
   @Test
   public void testKafkaWriteSchemaTransformConfigurationSchema() throws NoSuchSchemaException {
     Schema schema =