docs: @Override
  	public void open(Configuration configuration) throws Exception {
  		if (logFailuresOnly) {
  			callback = new Callback() {
  				@Override
  				public void onCompletion(RecordMetadata metadata, Exception e) {
  					if (e != null) {
  						LOG.error("Error while sending record to Kafka: " + e.getMessage(), e);
  					}
  					acknowledgeMessage();
  				}
  			};
  		}
  		else {
  			callback = new Callback() {
  				@Override
  				public void onCompletion(RecordMetadata metadata, Exception exception) {
  					if (exception != null && asyncException == null) {
  						asyncException = exception;
  					}
  					acknowledgeMessage();
  				}
  			};
  		}
  
  		super.open(configuration);
  	}