Skip to content

Add EnhancedLogFactory #1054

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 2 commits into from
Closed

Add EnhancedLogFactory #1054

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e27bb22
Add EnhancedLogFactory
garyrussell Mar 29, 2019
5d21fb8
Use Supplier for all conditional logging.
garyrussell Apr 1, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 20 additions & 28 deletions spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,8 +30,6 @@
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
Expand All @@ -52,6 +50,8 @@
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextStoppedEvent;
import org.springframework.kafka.support.EnhancedLogFactory;
import org.springframework.kafka.support.EnhancedLogFactory.Log;
import org.springframework.kafka.support.TransactionSupport;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
Expand Down Expand Up @@ -92,7 +92,7 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
*/
public static final Duration DEFAULT_PHYSICAL_CLOSE_TIMEOUT = Duration.ofSeconds(30);

private static final Log logger = LogFactory.getLog(DefaultKafkaProducerFactory.class); // NOSONAR
private static final Log LOGGER = EnhancedLogFactory.getLog(DefaultKafkaProducerFactory.class);

private final Map<String, Object> configs;

Expand Down Expand Up @@ -177,8 +177,8 @@ public void setTransactionIdPrefix(String transactionIdPrefix) {
*/
private void enableIdempotentBehaviour() {
Object previousValue = this.configs.putIfAbsent(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
if (logger.isDebugEnabled() && Boolean.FALSE.equals(previousValue)) {
logger.debug("The '" + ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG +
if (LOGGER.isDebugEnabled() && Boolean.FALSE.equals(previousValue)) {
LOGGER.debug("The '" + ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG +
"' is set to false, may result in duplicate messages");
}
}
Expand Down Expand Up @@ -233,7 +233,7 @@ public void destroy() {
producerToClose.delegate.close(this.physicalCloseTimeout);
}
catch (Exception e) {
logger.error("Exception while closing producer", e);
LOGGER.error("Exception while closing producer", e);
}
producerToClose = this.cache.poll();
}
Expand Down Expand Up @@ -280,7 +280,7 @@ public void reset() {
destroy();
}
catch (Exception e) {
logger.error("Exception while closing producer", e);
LOGGER.error("Exception while closing producer", e);
}
}

Expand Down Expand Up @@ -441,16 +441,19 @@ protected static class CloseSafeProducer<K, V> implements Producer<K, V> {

@Override
public Future<RecordMetadata> send(ProducerRecord<K, V> record) {
LOGGER.trace(() -> toString() + " send(" + record + ")");
return this.delegate.send(record);
}

@Override
public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
LOGGER.trace(() -> toString() + " send(" + record + ")");
return this.delegate.send(record, callback);
}

@Override
public void flush() {
LOGGER.trace(() -> toString() + " flush()");
this.delegate.flush();
}

Expand All @@ -471,16 +474,12 @@ public void initTransactions() {

@Override
public void beginTransaction() throws ProducerFencedException {
if (logger.isDebugEnabled()) {
logger.debug("beginTransaction: " + this);
}
LOGGER.debug(() -> toString() + " beginTransaction()");
try {
this.delegate.beginTransaction();
}
catch (RuntimeException e) {
if (logger.isErrorEnabled()) {
logger.error("beginTransaction failed: " + this, e);
}
LOGGER.error(() -> "beginTransaction failed: " + this, e);
this.txFailed = true;
throw e;
}
Expand All @@ -490,38 +489,31 @@ public void beginTransaction() throws ProducerFencedException {
public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets, String consumerGroupId)
throws ProducerFencedException {

LOGGER.trace(() -> toString() + " sendOffsetsToTransaction(" + offsets + ", " + consumerGroupId + ")");
this.delegate.sendOffsetsToTransaction(offsets, consumerGroupId);
}

@Override
public void commitTransaction() throws ProducerFencedException {
if (logger.isDebugEnabled()) {
logger.debug("commitTransaction: " + this);
}
LOGGER.debug(() -> toString() + " commitTransaction()");
try {
this.delegate.commitTransaction();
}
catch (RuntimeException e) {
if (logger.isErrorEnabled()) {
logger.error("commitTransaction failed: " + this, e);
}
LOGGER.error(() -> "commitTransaction failed: " + this, e);
this.txFailed = true;
throw e;
}
}

@Override
public void abortTransaction() throws ProducerFencedException {
if (logger.isDebugEnabled()) {
logger.debug("abortTransaction: " + this);
}
LOGGER.debug(() -> toString() + " abortTransaction()");
try {
this.delegate.abortTransaction();
}
catch (RuntimeException e) {
if (logger.isErrorEnabled()) {
logger.error("Abort failed: " + this, e);
}
LOGGER.error(() -> "Abort failed: " + this, e);
this.txFailed = true;
throw e;
}
Expand All @@ -541,12 +533,12 @@ public void close(long timeout, @Nullable TimeUnit unit) {

@Override
public void close(@Nullable Duration timeout) {
LOGGER.trace(() -> toString() + " close(" + (timeout == null ? "null" : timeout) + ")");
if (this.cache != null) {
if (this.txFailed) {
if (logger.isWarnEnabled()) {
logger.warn("Error during transactional operation; producer removed from cache; possible cause: "
LOGGER.warn(() -> "Error during transactional operation; producer removed from cache; "
+ "possible cause: "
+ "broker restarted during transaction: " + this);
}
if (timeout == null) {
this.delegate.close();
}
Expand Down
211 changes: 211 additions & 0 deletions spring-kafka/src/main/java/org/springframework/kafka/support/EnhancedLogFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,211 @@
/*
* Copyright 2019 the original author or authors.
*
* Licensed 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
*
* https://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.springframework.kafka.support;

import java.util.function.Supplier;

import org.apache.commons.logging.LogFactory;

/**
* Commons logging wrapper with {@link Supplier} log methods.
*
* @author Gary Russell
* @since 2.3
*
*/
public final class EnhancedLogFactory {

private EnhancedLogFactory() {
super();
}

/**
* Convenience method to return a named logger.
* @param clazz containing Class from which a log name will be derived
* @return the log.
*/
public static Log getLog(Class<?> clazz) {
return getLog(clazz.getName());
}

/**
* Convenience method to return a named logger.
* @param name logical name of the <code>Log</code> instance to be returned
* @return the log.
*/
public static Log getLog(String name) {
return new Log(LogFactory.getLog(name));
}

/**
* Wrapper for {@link Log} with {@link Supplier} methods.
*/
public static class Log {

private final org.apache.commons.logging.Log log;

public Log(org.apache.commons.logging.Log log) {
this.log = log;
}

public boolean isFatalEnabled() {
return this.log.isFatalEnabled();
}

public boolean isErrorEnabled() {
return this.log.isErrorEnabled();
}

public boolean isWarnEnabled() {
return this.log.isWarnEnabled();
}

public boolean isInfoEnabled() {
return this.log.isInfoEnabled();
}

public boolean isDebugEnabled() {
return this.log.isDebugEnabled();
}

public boolean isTraceEnabled() {
return this.log.isTraceEnabled();
}

public void fatal(Object message) {
this.log.fatal(message);
}

public void fatal(Object message, Throwable t) {
this.log.fatal(message, t);
}

public void error(Object message) {
this.log.error(message);
}

public void error(Object message, Throwable t) {
this.log.error(message, t);
}

public void warn(Object message) {
this.log.warn(message);
}

public void warn(Object message, Throwable t) {
this.log.warn(message, t);
}

public void info(Object message) {
this.log.info(message);
}

public void info(Object message, Throwable t) {
this.log.info(message, t);
}

public void debug(Object message) {
this.log.debug(message);
}

public void debug(Object message, Throwable t) {
this.log.debug(message, t);
}

public void trace(Object message) {
this.log.trace(message);
}

public void trace(Object message, Throwable t) {
this.log.trace(message, t);
}

public void fatal(Supplier<Object> supplier) {
if (this.log.isFatalEnabled()) {
this.log.fatal(supplier.get());
}
}

public void fatal(Supplier<Object> supplier, Throwable t) {
if (this.log.isFatalEnabled()) {
this.log.fatal(supplier.get(), t);
}
}

public void error(Supplier<Object> supplier) {
if (this.log.isErrorEnabled()) {
this.log.error(supplier.get());
}
}

public void error(Supplier<Object> supplier, Throwable t) {
if (this.log.isErrorEnabled()) {
this.log.error(supplier.get(), t);
}
}

public void warn(Supplier<Object> supplier) {
if (this.log.isWarnEnabled()) {
this.log.warn(supplier.get());
}
}

public void warn(Supplier<Object> supplier, Throwable t) {
if (this.log.isWarnEnabled()) {
this.log.warn(supplier.get(), t);
}
}

public void info(Supplier<Object> supplier) {
if (this.log.isInfoEnabled()) {
this.log.info(supplier.get());
}
}

public void info(Supplier<Object> supplier, Throwable t) {
if (this.log.isInfoEnabled()) {
this.log.info(supplier.get(), t);
}
}

public void debug(Supplier<Object> supplier) {
if (this.log.isDebugEnabled()) {
this.log.debug(supplier.get());
}
}

public void debug(Supplier<Object> supplier, Throwable t) {
if (this.log.isDebugEnabled()) {
this.log.debug(supplier.get(), t);
}
}

public void trace(Supplier<Object> supplier) {
if (this.log.isTraceEnabled()) {
this.log.trace(supplier.get());
}
}

public void trace(Supplier<Object> supplier, Throwable t) {
if (this.log.isTraceEnabled()) {
this.log.trace(supplier.get(), t);
}
}

}

}