content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements. See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You 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
   *
   *    http://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.apache.kafka.tools.consumer;
  
  import org.apache.kafka.clients.consumer.ConsumerRecord;
  import org.apache.kafka.common.MessageFormatter;
  import org.apache.kafka.common.header.Header;
  import org.apache.kafka.common.record.TimestampType;
  import org.apache.kafka.common.serialization.Deserializer;
  
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.io.IOException;
  import java.io.PrintStream;
  import java.nio.charset.StandardCharsets;
  import java.util.HashMap;
  import java.util.Iterator;
  import java.util.Map;
  import java.util.Optional;
  
  class DefaultMessageFormatter implements MessageFormatter {
  
      private static final Logger LOG = LoggerFactory.getLogger(DefaultMessageFormatter.class);
  
      private boolean printTimestamp = false;
      private boolean printKey = false;
      private boolean printValue = true;
      private boolean printPartition = false;
      private boolean printOffset = false;
      private boolean printDelivery = false;
      private boolean printEpoch = false;
      private boolean printHeaders = false;
      private byte[] keySeparator = utfBytes("\t");
      private byte[] lineSeparator = utfBytes("\n");
      private byte[] headersSeparator = utfBytes(",");
      private byte[] nullLiteral = utfBytes("null");
  
      private Optional<Deserializer<?>> keyDeserializer = Optional.empty();
      private Optional<Deserializer<?>> valueDeserializer = Optional.empty();
      private Optional<Deserializer<?>> headersDeserializer = Optional.empty();
  
      @Override
      public void configure(Map<String, ?> configs) {
          if (configs.containsKey("print.timestamp")) {
              printTimestamp = getBoolProperty(configs, "print.timestamp");
          }
          if (configs.containsKey("print.key")) {
              printKey = getBoolProperty(configs, "print.key");
          }
          if (configs.containsKey("print.offset")) {
              printOffset = getBoolProperty(configs, "print.offset");
          }
          if (configs.containsKey("print.delivery")) {
              printDelivery = getBoolProperty(configs, "print.delivery");
          }
          if (configs.containsKey("print.epoch")) {
              printEpoch = getBoolProperty(configs, "print.epoch");
          }
          if (configs.containsKey("print.partition")) {
              printPartition = getBoolProperty(configs, "print.partition");
          }
          if (configs.containsKey("print.headers")) {
              printHeaders = getBoolProperty(configs, "print.headers");
          }
          if (configs.containsKey("print.value")) {
              printValue = getBoolProperty(configs, "print.value");
          }
          if (configs.containsKey("key.separator")) {
              keySeparator = getByteProperty(configs, "key.separator");
          }
          if (configs.containsKey("line.separator")) {
              lineSeparator = getByteProperty(configs, "line.separator");
          }
          if (configs.containsKey("headers.separator")) {
              headersSeparator = getByteProperty(configs, "headers.separator");
          }
          if (configs.containsKey("null.literal")) {
              nullLiteral = getByteProperty(configs, "null.literal");
          }
  
          keyDeserializer = getDeserializerProperty(configs, true, "key.deserializer");
          valueDeserializer = getDeserializerProperty(configs, false, "value.deserializer");
          headersDeserializer = getDeserializerProperty(configs, false, "headers.deserializer");
      }
  
      // for testing
      public boolean printValue() {
          return printValue;
      }
  
      // for testing
      public Optional<Deserializer<?>> keyDeserializer() {
          return keyDeserializer;
      }
  
      private void writeSeparator(PrintStream output, boolean columnSeparator) {
          try {
              if (columnSeparator) {
                  output.write(keySeparator);
              } else {
                  output.write(lineSeparator);
              }
          } catch (IOException ioe) {
              LOG.error("Unable to write the separator to the output", ioe);
          }
      }
  
      private byte[] deserialize(ConsumerRecord<byte[], byte[]> consumerRecord, Optional<Deserializer<?>> deserializer, byte[] sourceBytes, String topic) {
          byte[] nonNullBytes = sourceBytes != null ? sourceBytes : nullLiteral;
          return deserializer.map(value -> utfBytes(value.deserialize(topic, consumerRecord.headers(), nonNullBytes).toString())).orElse(nonNullBytes);
      }
  
      @Override
      public void writeTo(ConsumerRecord<byte[], byte[]> consumerRecord, PrintStream output) {
          try {
              if (printTimestamp) {
                  if (consumerRecord.timestampType() != TimestampType.NO_TIMESTAMP_TYPE) {
                      output.print(consumerRecord.timestampType() + ":" + consumerRecord.timestamp());
                  } else {
                      output.print("NO_TIMESTAMP");
                  }
                  writeSeparator(output, printPartition || printOffset || printDelivery || printEpoch || printHeaders || printKey || printValue);
              }
  
              if (printPartition) {
                  output.print("Partition:");
                  output.print(consumerRecord.partition());
                  writeSeparator(output, printOffset || printDelivery || printEpoch || printHeaders || printKey || printValue);
              }
  
              if (printOffset) {
                  output.print("Offset:");
                  output.print(consumerRecord.offset());
                  writeSeparator(output, printDelivery || printEpoch || printHeaders || printKey || printValue);
              }
  
              if (printDelivery) {
                  output.print("Delivery:");
                  output.print(consumerRecord.deliveryCount().map(delivery -> Short.toString(delivery)).orElse("NOT_PRESENT"));
                  writeSeparator(output, printEpoch || printHeaders || printKey || printValue);
              }
  
              if (printEpoch) {
                  output.print("Epoch:");
                  output.print(consumerRecord.leaderEpoch().map(epoch -> Integer.toString(epoch)).orElse("NOT_PRESENT"));
                  writeSeparator(output, printHeaders || printKey || printValue);
              }
  
              if (printHeaders) {
                  Iterator<Header> headersIt = consumerRecord.headers().iterator();
                  if (!headersIt.hasNext()) {
                      output.print("NO_HEADERS");
                  } else {
                      while (headersIt.hasNext()) {
                          Header header = headersIt.next();
                          output.print(header.key() + ":");
                          output.write(deserialize(consumerRecord, headersDeserializer, header.value(), consumerRecord.topic()));
                          if (headersIt.hasNext()) {
                              output.write(headersSeparator);
                          }
                      }
                  }
                  writeSeparator(output, printKey || printValue);
              }
  
              if (printKey) {
                  output.write(deserialize(consumerRecord, keyDeserializer, consumerRecord.key(), consumerRecord.topic()));
                  writeSeparator(output, printValue);
              }
  
              if (printValue) {
                  output.write(deserialize(consumerRecord, valueDeserializer, consumerRecord.value(), consumerRecord.topic()));
                  output.write(lineSeparator);
              }
          } catch (IOException ioe) {
              LOG.error("Unable to write the consumer record to the output", ioe);
          }
      }
  
      private Map<String, ?> propertiesWithKeyPrefixStripped(String prefix, Map<String, ?> configs) {
          final Map<String, Object> newConfigs = new HashMap<>();
          for (Map.Entry<String, ?> entry : configs.entrySet()) {
              if (entry.getKey().startsWith(prefix) && entry.getKey().length() > prefix.length()) {
                  newConfigs.put(entry.getKey().substring(prefix.length()), entry.getValue());
              }
          }
          return newConfigs;
      }
  
      private byte[] utfBytes(String str) {
          return str.getBytes(StandardCharsets.UTF_8);
      }
  
      private byte[] getByteProperty(Map<String, ?> configs, String key) {
          return utfBytes((String) configs.get(key));
      }
  
      private boolean getBoolProperty(Map<String, ?> configs, String key) {
          return ((String) configs.get(key)).trim().equalsIgnoreCase("true");
      }
  
      private Optional<Deserializer<?>> getDeserializerProperty(Map<String, ?> configs, boolean isKey, String key) {
          if (configs.containsKey(key)) {
              String name = (String) configs.get(key);
              try {
                  Deserializer<?> deserializer = (Deserializer<?>) Class.forName(name).getDeclaredConstructor().newInstance();
                  Map<String, ?> deserializerConfig = propertiesWithKeyPrefixStripped(key + ".", configs);
                  deserializer.configure(deserializerConfig, isKey);
                  return Optional.of(deserializer);
              } catch (Exception e) {
                  LOG.error("Unable to instantiate a deserializer from " + name, e);
              }
          }
          return Optional.empty();
      }
  }
  