MyOwnErrorCallback

package test;

/**
 * Created by rahuljai on 5/7/15.
 */
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class MyOwnErrorCallback implements Callback {
    private String topic;
    private byte[] key;
    private byte[] value;
    private boolean logAsString;

    public MyOwnErrorCallback(String topic, byte[] key, byte[] value, boolean logAsString) {
        this.topic = topic;
        this.key = key;
        this.value = value;
        this.logAsString = logAsString;
    }

    public void onCompletion(RecordMetadata metadata, Exception e) {
        if(e != null) {
            String keyString = this.key == null?"null":(this.logAsString?new String(this.key):this.key.length + " bytes");
            String valueString = this.value == null?"null":(this.logAsString?new String(this.value):this.value.length + " bytes");


            String msg = String.format("Error when sending message to topic %s with key: %s, value: %s with error: %s", new Object[]{this.topic, keyString, valueString, e.getMessage()});

            System.out.println(msg);
        }

    }
}