diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/AssignedTasks.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/AssignedTasks.java index 9204571314672..4c4941a2992b6 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/AssignedTasks.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/AssignedTasks.java @@ -94,7 +94,7 @@ void initializeNewTasks() { it.remove(); } catch (final LockException e) { // made this trace as it will spam the logs in the poll loop. - log.trace("Could not create {} {} due to {}; will retry", taskTypeName, entry.getKey(), e.getMessage()); + log.trace("Could not create {} {} due to {}; will retry", taskTypeName, entry.getKey(), e.toString()); } } } @@ -204,7 +204,7 @@ RuntimeException closeZombieTask(final T task) { try { task.close(false, true); } catch (final RuntimeException e) { - log.warn("Failed to close zombie {} {} due to {}; ignore and proceed.", taskTypeName, task.id(), e.getMessage()); + log.warn("Failed to close zombie {} {} due to {}; ignore and proceed.", taskTypeName, task.id(), e.toString()); return e; } return null; diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java index 56e6bed08501d..e8ec5e9fe5f7a 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java @@ -276,7 +276,7 @@ private void restoreState(final StateRestoreCallback stateRestoreCallback, } catch (final InvalidOffsetException recoverableException) { log.warn("Restoring GlobalStore {} failed due to: {}. Deleting global store to recreate from scratch.", storeName, - recoverableException.getMessage()); + recoverableException.toString()); reinitializeStateStoresForPartitions(recoverableException.partitions(), processorContext); stateRestoreListener.onRestoreStart(topicPartition, storeName, offset, highWatermark); @@ -318,7 +318,7 @@ public void close(final Map offsets) throws IOException { closeFailed.append("Failed to close global state store:") .append(entry.getKey()) .append(". Reason: ") - .append(e.getMessage()) + .append(e.toString()) .append("\n"); } } diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java index aeff946ce1efb..2ac37bdad20f7 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java @@ -130,9 +130,9 @@ public void makeReady(final Map topics) { "Will try again (remaining retries {}).", topicName, remainingRetries - 1); } else if (cause instanceof TopicExistsException) { createTopicNames.add(createTopicResult.getKey()); - log.info(String.format("Topic %s exist already: %s", + log.info("Topic {} exist already: {}", topicName, - couldNotCreateTopic.getMessage())); + couldNotCreateTopic.toString()); } else { throw new StreamsException(String.format("Could not create topic %s.", topicName), couldNotCreateTopic); @@ -197,8 +197,8 @@ protected Map getNumPartitions(final Set topics) { log.debug("Could not get number of partitions for topic {} due to timeout. " + "Will try again (remaining retries {}).", topicFuture.getKey(), remainingRetries - 1); } else { - final String error = "Could not get number of partitions for topic {}."; - log.debug(error, topicFuture.getKey(), cause.getMessage()); + final String error = "Could not get number of partitions for topic {} due to {}"; + log.debug(error, topicFuture.getKey(), cause.toString()); } } } diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java index 286cd817c88be..f56804824fb5e 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java @@ -130,7 +130,7 @@ private void recordSendError( value, timestamp, topic, - exception.getMessage()), + exception.toString()), exception); } @@ -163,7 +163,7 @@ public void onCompletion(final RecordMetadata metadata, } else { if (sendException == null) { if (exception instanceof ProducerFencedException) { - log.warn(LOG_MESSAGE, key, value, timestamp, topic, exception.getMessage()); + log.warn(LOG_MESSAGE, key, value, timestamp, topic, exception.toString()); sendException = new ProducerFencedException( String.format(EXCEPTION_MESSAGE, logPrefix, @@ -172,7 +172,7 @@ public void onCompletion(final RecordMetadata metadata, value, timestamp, topic, - exception.getMessage())); + exception.toString())); } else { if (productionExceptionIsFatal(exception)) { recordSendError(key, value, timestamp, topic, exception); @@ -201,7 +201,7 @@ public void onCompletion(final RecordMetadata metadata, value, timestamp, topic, - uncaughtException.getMessage()), + uncaughtException.toString()), uncaughtException); } } diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java index 02a4bb92137e8..ab96cce4c2fea 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java @@ -260,7 +260,7 @@ public void onPartitionsAssigned(final Collection assignment) { taskManager.createTasks(assignment); } catch (final Throwable t) { log.error("Error caught during partition assignment, " + - "will abort the current process and re-throw at the end of rebalance: {}", t.getMessage()); + "will abort the current process and re-throw at the end of rebalance: {}", t); streamThread.setRebalanceException(t); } finally { log.info("partition assignment took {} ms.\n" + @@ -291,7 +291,7 @@ public void onPartitionsRevoked(final Collection assignment) { taskManager.suspendTasksAndState(); } catch (final Throwable t) { log.error("Error caught during partition revocation, " + - "will abort the current process and re-throw at the end of rebalance: {}", t.getMessage()); + "will abort the current process and re-throw at the end of rebalance: {}", t); streamThread.setRebalanceException(t); } finally { streamThread.clearStandbyRecords();