17
17
import org .apache .kafka .connect .header .ConnectHeaders ;
18
18
import org .apache .kafka .connect .header .Headers ;
19
19
import org .apache .kafka .connect .runtime .ConnectorConfig ;
20
- import org .apache .kafka .connect .runtime .errors .DeadLetterQueueReporter ;
21
20
import org .apache .kafka .connect .runtime .errors .ToleranceType ;
22
21
import org .apache .kafka .connect .source .SourceRecord ;
23
22
import org .slf4j .Logger ;
@@ -35,12 +34,6 @@ public class ErrorHandler {
35
34
36
35
public static final String HEADER_PREFIX = "__connect.errors." ;
37
36
public static final String ERROR_HEADER_ORIG_TOPIC = HEADER_PREFIX + "topic" ;
38
- public static final String ERROR_HEADER_ORIG_PARTITION = HEADER_PREFIX + "partition" ;
39
- public static final String ERROR_HEADER_ORIG_OFFSET = HEADER_PREFIX + "offset" ;
40
- public static final String ERROR_HEADER_CONNECTOR_NAME = HEADER_PREFIX + "connector.name" ;
41
- public static final String ERROR_HEADER_TASK_ID = HEADER_PREFIX + "task.id" ;
42
- public static final String ERROR_HEADER_STAGE = HEADER_PREFIX + "stage" ;
43
- public static final String ERROR_HEADER_EXECUTING_CLASS = HEADER_PREFIX + "class.name" ;
44
37
public static final String ERROR_HEADER_EXCEPTION = HEADER_PREFIX + "exception.class.name" ;
45
38
public static final String ERROR_HEADER_EXCEPTION_MESSAGE = HEADER_PREFIX + "exception.message" ;
46
39
public static final String ERROR_HEADER_EXCEPTION_STACK_TRACE = HEADER_PREFIX + "exception.stacktrace" ;
@@ -113,7 +106,6 @@ private void initializeErrorTolerance(final Map<String, String> props) {
113
106
dlqTopic = props .get (MQSourceConnector .DLQ_TOPIC_NAME_CONFIG );
114
107
if (dlqTopic != null && !dlqTopic .isEmpty ()) {
115
108
dlqTopic = dlqTopic .trim ();
116
- // TODO: Check if DLQ topic exists
117
109
}
118
110
119
111
queueName = props .get (MQSourceConnector .CONFIG_NAME_MQ_QUEUE );
@@ -282,9 +274,9 @@ private Headers createErrorHeaders(final Message message, final String originalT
282
274
}
283
275
284
276
// Basic error information
285
- headers .addString (DeadLetterQueueReporter . ERROR_HEADER_ORIG_TOPIC , originalTopic );
286
- headers .addString (DeadLetterQueueReporter . ERROR_HEADER_EXECUTING_CLASS , exception .getClass ().getName ());
287
- headers .addString (DeadLetterQueueReporter . ERROR_HEADER_EXCEPTION_MESSAGE , exception .getMessage ());
277
+ headers .addString (ERROR_HEADER_ORIG_TOPIC , originalTopic );
278
+ headers .addString (ERROR_HEADER_EXCEPTION , exception .getClass ().getName ());
279
+ headers .addString (ERROR_HEADER_EXCEPTION_MESSAGE , exception .getMessage ());
288
280
289
281
try {
290
282
headers .addString (ERROR_HEADER_JMS_MESSAGE_ID , message .getJMSMessageID ());
@@ -298,14 +290,14 @@ private Headers createErrorHeaders(final Message message, final String originalT
298
290
299
291
// Add cause if available
300
292
if (exception .getCause () != null ) {
301
- headers .addString (ERROR_HEADER_EXCEPTION_CAUSE_MESSAGE , exception .getCause ().getMessage ());
302
293
headers .addString (ERROR_HEADER_EXCEPTION_CAUSE_CLASS , exception .getCause ().getClass ().getName ());
294
+ headers .addString (ERROR_HEADER_EXCEPTION_CAUSE_MESSAGE , exception .getCause ().getMessage ());
303
295
}
304
296
305
297
// Add first few lines of stack trace (full stack trace might be too large)
306
298
final String stackTrace = getStackTrace (exception );
307
299
if (stackTrace != null ) {
308
- headers .addString (DeadLetterQueueReporter . ERROR_HEADER_EXCEPTION_STACK_TRACE , stackTrace );
300
+ headers .addString (ERROR_HEADER_EXCEPTION_STACK_TRACE , stackTrace );
309
301
}
310
302
311
303
return headers ;
0 commit comments