From 2acb03a2d192baa27775c24e9dc576d2ab981195 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Lo=C3=AFc=20Greffier?= <loic.greffier@michelin.com>
Date: Mon, 27 May 2024 17:06:02 +0200
Subject: [PATCH] KAFKA-16448: Update ErrorHandlerContext

Co-authored-by: Dabz <d.gasparina@gmail.com>
Co-authored-by: sebastienviale <sebastien.viale@michelin.com>
---
 .../streams/errors/ErrorHandlerContext.java   |  6 --
 .../errors/ErrorHandlerContextImpl.java       | 90 +++++++++++++++++++
 2 files changed, 90 insertions(+), 6 deletions(-)
 create mode 100644 streams/src/main/java/org/apache/kafka/streams/errors/ErrorHandlerContextImpl.java

diff --git a/streams/src/main/java/org/apache/kafka/streams/errors/ErrorHandlerContext.java b/streams/src/main/java/org/apache/kafka/streams/errors/ErrorHandlerContext.java
index c7ac5fbc45957..5f28b711f2e1c 100644
--- a/streams/src/main/java/org/apache/kafka/streams/errors/ErrorHandlerContext.java
+++ b/streams/src/main/java/org/apache/kafka/streams/errors/ErrorHandlerContext.java
@@ -99,9 +99,6 @@ public interface ErrorHandlerContext {
      * <p> If this method is invoked in a sub-topology due to a repartition, the returned key would be one sent
      * to the repartition topic.
      *
-     * <p> Always returns null if this method is invoked within a
-     * {@link org.apache.kafka.streams.errors.ProductionExceptionHandler#handle(ErrorHandlerContext, org.apache.kafka.clients.producer.ProducerRecord, Exception)}
-     *
      * @return the raw byte of the key of the source message
      */
     byte[] sourceRawKey();
@@ -116,9 +113,6 @@ public interface ErrorHandlerContext {
      * <p> If this method is invoked in a sub-topology due to a repartition, the returned value would be one sent
      * to the repartition topic.
      *
-     * <p> Always returns null if this method is invoked within a
-     * {@link org.apache.kafka.streams.errors.ProductionExceptionHandler#handle(ErrorHandlerContext, org.apache.kafka.clients.producer.ProducerRecord, Exception)}
-     *
      * @return the raw byte of the value of the source message
      */
     byte[] sourceRawValue();
diff --git a/streams/src/main/java/org/apache/kafka/streams/errors/ErrorHandlerContextImpl.java b/streams/src/main/java/org/apache/kafka/streams/errors/ErrorHandlerContextImpl.java
new file mode 100644
index 0000000000000..c801799a6cdad
--- /dev/null
+++ b/streams/src/main/java/org/apache/kafka/streams/errors/ErrorHandlerContextImpl.java
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.errors;
+
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.streams.processor.TaskId;
+
+public class ErrorHandlerContextImpl implements ErrorHandlerContext {
+    private final String topic;
+    private final int partition;
+    private final long offset;
+    private final Headers headers;
+    private final byte[] sourceRawKey;
+    private final byte[] sourceRawValue;
+    private final String processorNodeId;
+    private final TaskId taskId;
+
+    public ErrorHandlerContextImpl(final String topic,
+        final int partition,
+        final long offset,
+        final Headers headers,
+        final byte[] sourceRawKey,
+        final byte[] sourceRawValue,
+        final String processorNodeId,
+        final TaskId taskId) {
+        this.topic = topic;
+        this.partition = partition;
+        this.offset = offset;
+        this.headers = headers;
+        this.sourceRawKey = sourceRawKey;
+        this.sourceRawValue = sourceRawValue;
+        this.processorNodeId = processorNodeId;
+        this.taskId = taskId;
+    }
+
+
+    @Override
+    public String topic() {
+        return this.topic;
+    }
+
+    @Override
+    public int partition() {
+        return this.partition;
+    }
+
+    @Override
+    public long offset() {
+        return this.offset;
+    }
+
+    @Override
+    public Headers headers() {
+        return this.headers;
+    }
+
+    @Override
+    public byte[] sourceRawKey() {
+        return this.sourceRawKey;
+    }
+
+    @Override
+    public byte[] sourceRawValue() {
+        return this.sourceRawValue;
+    }
+
+    @Override
+    public String processorNodeId() {
+        return this.processorNodeId;
+    }
+
+    @Override
+    public TaskId taskId() {
+        return this.taskId;
+    }
+}