kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jkr...@apache.org
Subject [5/6] kafka git commit: KAFKA-1915: Add checkstyle for java code.
Date Wed, 04 Feb 2015 04:58:44 GMT
http://git-wip-us.apache.org/repos/asf/kafka/blob/1c6d5bba/clients/src/main/java/org/apache/kafka/common/message/KafkaLZ4BlockInputStream.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/message/KafkaLZ4BlockInputStream.java
b/clients/src/main/java/org/apache/kafka/common/message/KafkaLZ4BlockInputStream.java
deleted file mode 100644
index 5be72fe..0000000
--- a/clients/src/main/java/org/apache/kafka/common/message/KafkaLZ4BlockInputStream.java
+++ /dev/null
@@ -1,233 +0,0 @@
-/**
- * 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.common.message;
-
-import static org.apache.kafka.common.message.KafkaLZ4BlockOutputStream.LZ4_FRAME_INCOMPRESSIBLE_MASK;
-import static org.apache.kafka.common.message.KafkaLZ4BlockOutputStream.LZ4_MAX_HEADER_LENGTH;
-import static org.apache.kafka.common.message.KafkaLZ4BlockOutputStream.MAGIC;
-
-import java.io.FilterInputStream;
-import java.io.IOException;
-import java.io.InputStream;
-
-import org.apache.kafka.common.message.KafkaLZ4BlockOutputStream.BD;
-import org.apache.kafka.common.message.KafkaLZ4BlockOutputStream.FLG;
-import org.apache.kafka.common.utils.Utils;
-
-import net.jpountz.lz4.LZ4Exception;
-import net.jpountz.lz4.LZ4Factory;
-import net.jpountz.lz4.LZ4SafeDecompressor;
-import net.jpountz.xxhash.XXHash32;
-import net.jpountz.xxhash.XXHashFactory;
-
-/**
- * A partial implementation of the v1.4.1 LZ4 Frame format.
- * 
- * @see <a href="https://docs.google.com/document/d/1Tdxmn5_2e5p1y4PtXkatLndWVb0R8QARJFe6JI4Keuo/edit">LZ4
Framing Format Spec</a>
- */
-public final class KafkaLZ4BlockInputStream extends FilterInputStream {
-
-  public static final String PREMATURE_EOS = "Stream ended prematurely";
-  public static final String NOT_SUPPORTED = "Stream unsupported";
-  public static final String BLOCK_HASH_MISMATCH = "Block checksum mismatch";
-  public static final String DESCRIPTOR_HASH_MISMATCH = "Stream frame descriptor corrupted";
-  
-  private final LZ4SafeDecompressor decompressor;
-  private final XXHash32 checksum;
-  private final byte[] buffer;
-  private final byte[] compressedBuffer;
-  private final int maxBlockSize;
-  private FLG flg;
-  private BD bd;
-  private int bufferOffset;
-  private int bufferSize;
-  private boolean finished;
-  
-  /**
-   * Create a new {@link InputStream} that will decompress data using the LZ4 algorithm.
-   * 
-   * @param in The stream to decompress
-   * @throws IOException
-   */
-  public KafkaLZ4BlockInputStream(InputStream in) throws IOException {
-    super(in);
-    decompressor = LZ4Factory.fastestInstance().safeDecompressor();
-    checksum = XXHashFactory.fastestInstance().hash32();
-    readHeader();
-    maxBlockSize = bd.getBlockMaximumSize();
-    buffer = new byte[maxBlockSize];
-    compressedBuffer = new byte[maxBlockSize];
-    bufferOffset = 0;
-    bufferSize = 0;
-    finished = false;
-  }
-  
-  /**
-   * Reads the magic number and frame descriptor from the underlying {@link InputStream}.
-   *  
-   * @throws IOException
-   */
-  private void readHeader() throws IOException {
-    byte[] header = new byte[LZ4_MAX_HEADER_LENGTH];
-    
-    // read first 6 bytes into buffer to check magic and FLG/BD descriptor flags
-    bufferOffset = 6;
-    if (in.read(header, 0, bufferOffset) != bufferOffset) {
-      throw new IOException(PREMATURE_EOS);
-    }
-
-    if (MAGIC != Utils.readUnsignedIntLE(header, bufferOffset-6)) {
-      throw new IOException(NOT_SUPPORTED);
-    }
-    flg = FLG.fromByte(header[bufferOffset-2]);
-    bd = BD.fromByte(header[bufferOffset-1]);
-    // TODO read uncompressed content size, update flg.validate()
-    // TODO read dictionary id, update flg.validate()
-    
-    // check stream descriptor hash
-    byte hash = (byte) ((checksum.hash(header, 0, bufferOffset, 0) >> 8) & 0xFF);
-    header[bufferOffset++] = (byte) in.read();
-    if (hash != header[bufferOffset-1]) {
-      throw new IOException(DESCRIPTOR_HASH_MISMATCH);
-    }
-  }
-
-  /**
-   * Decompresses (if necessary) buffered data, optionally computes and validates a XXHash32
checksum, 
-   * and writes the result to a buffer.
-   * 
-   * @throws IOException
-   */
-  private void readBlock() throws IOException {
-    int blockSize = Utils.readUnsignedIntLE(in);
-
-    // Check for EndMark
-    if (blockSize == 0) {
-      finished = true;
-      // TODO implement content checksum, update flg.validate()
-      return;
-    } else if (blockSize > maxBlockSize) {
-      throw new IOException(String.format("Block size %s exceeded max: %s", blockSize, maxBlockSize));
-    }
-    
-    boolean compressed = (blockSize & LZ4_FRAME_INCOMPRESSIBLE_MASK) == 0;
-    byte[] bufferToRead;
-    if (compressed) {
-      bufferToRead = compressedBuffer;
-    } else {
-      blockSize &= ~LZ4_FRAME_INCOMPRESSIBLE_MASK;
-      bufferToRead = buffer;
-      bufferSize = blockSize;
-    }
-    
-    if (in.read(bufferToRead, 0, blockSize) != blockSize) {
-      throw new IOException(PREMATURE_EOS);
-    }
-
-    // verify checksum
-    if (flg.isBlockChecksumSet() && Utils.readUnsignedIntLE(in) != checksum.hash(bufferToRead,
0, blockSize, 0)) {
-      throw new IOException(BLOCK_HASH_MISMATCH);
-    }
-
-    if (compressed) {
-      try {
-        bufferSize = decompressor.decompress(compressedBuffer, 0, blockSize, buffer, 0, maxBlockSize);
-      } catch (LZ4Exception e) {
-        throw new IOException(e);
-      }
-    }
-
-    bufferOffset = 0;
-  }
- 
-  @Override
-  public int read() throws IOException {
-    if (finished) {
-      return -1;
-    }
-    if (available() == 0) {
-      readBlock();
-    }
-    if (finished) {
-      return -1;
-    }
-    int value = buffer[bufferOffset++] & 0xFF;
-    
-    return value;
-  }
-
-  @Override
-  public int read(byte b[], int off, int len) throws IOException {
-    net.jpountz.util.Utils.checkRange(b, off, len);
-    if (finished) {
-      return -1;
-    }
-    if (available() == 0) {
-      readBlock();
-    }
-    if (finished) {
-      return -1;
-    }
-    len = Math.min(len, available());
-    System.arraycopy(buffer, bufferOffset, b, off, len);
-    bufferOffset += len;
-    return len;
-  }
-
-  @Override
-  public long skip(long n) throws IOException {
-    if (finished) {
-      return 0;
-    }
-    if (available() == 0) {
-      readBlock();
-    }
-    if (finished) {
-      return 0;
-    }
-    n = Math.min(n, available());
-    bufferOffset += n;
-    return n;
-  }
-
-  @Override
-  public int available() throws IOException {
-    return bufferSize - bufferOffset;
-  }
-
-  @Override
-  public void close() throws IOException {
-      in.close();
-  }
-
-  @Override
-  public synchronized void mark(int readlimit) {
-      throw new RuntimeException("mark not supported");
-  }
-
-  @Override
-  public synchronized void reset() throws IOException {
-      throw new RuntimeException("reset not supported");
-  }
-
-  @Override
-  public boolean markSupported() {
-      return false;
-  }
-
-}

http://git-wip-us.apache.org/repos/asf/kafka/blob/1c6d5bba/clients/src/main/java/org/apache/kafka/common/message/KafkaLZ4BlockOutputStream.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/message/KafkaLZ4BlockOutputStream.java
b/clients/src/main/java/org/apache/kafka/common/message/KafkaLZ4BlockOutputStream.java
deleted file mode 100644
index e5b9e43..0000000
--- a/clients/src/main/java/org/apache/kafka/common/message/KafkaLZ4BlockOutputStream.java
+++ /dev/null
@@ -1,387 +0,0 @@
-/**
- * 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.common.message;
-
-import java.io.FilterOutputStream;
-import java.io.IOException;
-import java.io.OutputStream;
-
-import org.apache.kafka.common.utils.Utils;
-
-import net.jpountz.lz4.LZ4Compressor;
-import net.jpountz.lz4.LZ4Factory;
-import net.jpountz.xxhash.XXHash32;
-import net.jpountz.xxhash.XXHashFactory;
-
-/**
- * A partial implementation of the v1.4.1 LZ4 Frame format.
- * 
- * @see <a href="https://docs.google.com/document/d/1Tdxmn5_2e5p1y4PtXkatLndWVb0R8QARJFe6JI4Keuo/edit">LZ4
Framing Format Spec</a>
- */
-public final class KafkaLZ4BlockOutputStream extends FilterOutputStream {
-
-  public static final int MAGIC = 0x184D2204;
-  public static final int LZ4_MAX_HEADER_LENGTH = 19;
-  public static final int LZ4_FRAME_INCOMPRESSIBLE_MASK = 0x80000000;
-  
-  public static final String CLOSED_STREAM = "The stream is already closed";
-  
-  public static final int BLOCKSIZE_64KB = 4;
-  public static final int BLOCKSIZE_256KB = 5;
-  public static final int BLOCKSIZE_1MB = 6;
-  public static final int BLOCKSIZE_4MB = 7;
-  
-  private final LZ4Compressor compressor;
-  private final XXHash32 checksum;
-  private final FLG flg;
-  private final BD bd;
-  private final byte[] buffer;
-  private final byte[] compressedBuffer;
-  private final int maxBlockSize;
-  private int bufferOffset;
-  private boolean finished;
-
-  /**
-   * Create a new {@link OutputStream} that will compress data using the LZ4 algorithm.
-   *
-   * @param out The output stream to compress
-   * @param blockSize Default: 4. The block size used during compression. 4=64kb, 5=256kb,
6=1mb, 7=4mb. All other values will generate an exception
-   * @param blockChecksum Default: false. When true, a XXHash32 checksum is computed and
appended to the stream for every block of data
-   * @throws IOException
-   */
-  public KafkaLZ4BlockOutputStream(OutputStream out, int blockSize, boolean blockChecksum)
throws IOException {
-    super(out);
-    compressor = LZ4Factory.fastestInstance().fastCompressor();
-    checksum = XXHashFactory.fastestInstance().hash32();
-    bd = new BD(blockSize);
-    flg = new FLG(blockChecksum);
-    bufferOffset = 0;
-    maxBlockSize = bd.getBlockMaximumSize();
-    buffer = new byte[maxBlockSize];
-    compressedBuffer = new byte[compressor.maxCompressedLength(maxBlockSize)];
-    finished = false;
-    writeHeader();
-  }
-  
-  /**
-   * Create a new {@link OutputStream} that will compress data using the LZ4 algorithm.
-   *  
-   * @param out The stream to compress
-   * @param blockSize Default: 4. The block size used during compression. 4=64kb, 5=256kb,
6=1mb, 7=4mb. All other values will generate an exception
-   * @throws IOException
-   */
-  public KafkaLZ4BlockOutputStream(OutputStream out, int blockSize) throws IOException {
-    this(out, blockSize, false);
-  }
-  
-  /**
-   * Create a new {@link OutputStream} that will compress data using the LZ4 algorithm.
-   * 
-   * @param out The output stream to compress
-   * @throws IOException
-   */
-  public KafkaLZ4BlockOutputStream(OutputStream out) throws IOException {
-    this(out, BLOCKSIZE_64KB);
-  }
-
-  /**
-   * Writes the magic number and frame descriptor to the underlying {@link OutputStream}.
-   *  
-   * @throws IOException
-   */
-  private void writeHeader() throws IOException {
-    Utils.writeUnsignedIntLE(buffer, 0, MAGIC);
-    bufferOffset = 4;
-    buffer[bufferOffset++] = flg.toByte();
-    buffer[bufferOffset++] = bd.toByte();
-    // TODO write uncompressed content size, update flg.validate()
-    // TODO write dictionary id, update flg.validate()
-    // compute checksum on all descriptor fields 
-    int hash = (checksum.hash(buffer, 0, bufferOffset, 0) >> 8) & 0xFF;
-    buffer[bufferOffset++] = (byte) hash;
-    // write out frame descriptor
-    out.write(buffer, 0, bufferOffset);
-    bufferOffset = 0;
-  }
-  
-  /**
-   * Compresses buffered data, optionally computes an XXHash32 checksum, and writes
-   * the result to the underlying {@link OutputStream}.
-   * 
-   * @throws IOException
-   */
-  private void writeBlock() throws IOException {
-    if (bufferOffset == 0) {
-      return;
-    }
-    
-    int compressedLength = compressor.compress(buffer, 0, bufferOffset, compressedBuffer,
0);
-    byte[] bufferToWrite = compressedBuffer;
-    int compressMethod = 0;
-    
-    // Store block uncompressed if compressed length is greater (incompressible)
-    if (compressedLength >= bufferOffset) {
-      bufferToWrite = buffer;
-      compressedLength = bufferOffset;
-      compressMethod = LZ4_FRAME_INCOMPRESSIBLE_MASK;
-    }
-
-    // Write content
-    Utils.writeUnsignedIntLE(out, compressedLength | compressMethod);
-    out.write(bufferToWrite, 0, compressedLength);
-    
-    // Calculate and write block checksum
-    if (flg.isBlockChecksumSet()) {
-      int hash = checksum.hash(bufferToWrite, 0, compressedLength, 0);
-      Utils.writeUnsignedIntLE(out, hash);
-    }
-    bufferOffset = 0;
-  }
-  
-  /**
-   * Similar to the {@link #writeBlock()} method.  Writes a 0-length block 
-   * (without block checksum) to signal the end of the block stream.
-   * 
-   * @throws IOException
-   */
-  private void writeEndMark() throws IOException {
-    Utils.writeUnsignedIntLE(out, 0);
-    // TODO implement content checksum, update flg.validate()
-    finished = true;
-  }
-
-  @Override
-  public void write(int b) throws IOException {
-    ensureNotFinished();
-    if (bufferOffset == maxBlockSize) {
-      writeBlock();
-    }
-    buffer[bufferOffset++] = (byte) b;
-  }
-  
-  @Override
-  public void write(byte[] b, int off, int len) throws IOException {
-    net.jpountz.util.Utils.checkRange(b, off, len);
-    ensureNotFinished();
-     
-    int bufferRemainingLength = maxBlockSize - bufferOffset;
-    // while b will fill the buffer
-    while (len > bufferRemainingLength) {
-      // fill remaining space in buffer 
-      System.arraycopy(b, off, buffer, bufferOffset, bufferRemainingLength);
-      bufferOffset = maxBlockSize;
-      writeBlock();
-      // compute new offset and length
-      off += bufferRemainingLength;
-      len -= bufferRemainingLength;
-      bufferRemainingLength = maxBlockSize;
-    }
-    
-    System.arraycopy(b, off, buffer, bufferOffset, len);
-    bufferOffset += len;
-  }
-
-  @Override
-  public void flush() throws IOException {
-    if (!finished) {
-        writeBlock();
-    }
-    if (out != null) {
-      out.flush();
-    }
-  }
-
-  /**
-   * A simple state check to ensure the stream is still open.
-   */
-  private void ensureNotFinished() {
-    if (finished) {
-      throw new IllegalStateException(CLOSED_STREAM);
-    }
-  }
-
-  @Override
-  public void close() throws IOException {
-    if (!finished) {
-      writeEndMark();
-      flush();
-      finished = true;
-    }
-    if (out != null) {
-      out.close();
-      out = null;
-    }
-  }
-
-  public static class FLG {
-    
-    private static final int VERSION = 1;
-    
-    private final int presetDictionary;
-    private final int reserved1;
-    private final int contentChecksum;
-    private final int contentSize;
-    private final int blockChecksum;
-    private final int blockIndependence;
-    private final int version;
-    
-    public FLG() {
-      this(false);
-    }
-    
-    public FLG(boolean blockChecksum) {
-      this(0, 0, 0, 0, blockChecksum ? 1 : 0, 1, VERSION);
-    }
-    
-    private FLG(int presetDictionary, int reserved1, int contentChecksum, 
-        int contentSize, int blockChecksum, int blockIndependence, int version) {
-      this.presetDictionary = presetDictionary;
-      this.reserved1 = reserved1;
-      this.contentChecksum = contentChecksum;
-      this.contentSize = contentSize;
-      this.blockChecksum = blockChecksum;
-      this.blockIndependence = blockIndependence;
-      this.version = version;
-      validate();
-    }
-    
-    public static FLG fromByte(byte flg) {
-      int presetDictionary =  (flg >>> 0) & 1;
-      int reserved1 =         (flg >>> 1) & 1;
-      int contentChecksum =   (flg >>> 2) & 1;
-      int contentSize =       (flg >>> 3) & 1;
-      int blockChecksum =     (flg >>> 4) & 1;
-      int blockIndependence = (flg >>> 5) & 1;
-      int version =           (flg >>> 6) & 3;
-      
-      return new FLG(presetDictionary, reserved1, contentChecksum, 
-          contentSize, blockChecksum, blockIndependence, version);
-    }
-    
-    public byte toByte() {
-      return (byte) (
-            ((presetDictionary   & 1) << 0)
-          | ((reserved1          & 1) << 1)
-          | ((contentChecksum    & 1) << 2)
-          | ((contentSize        & 1) << 3)
-          | ((blockChecksum      & 1) << 4)
-          | ((blockIndependence  & 1) << 5)
-          | ((version            & 3) << 6) );
-    }
-    
-    private void validate() {
-      if (presetDictionary != 0) {
-        throw new RuntimeException("Preset dictionary is unsupported");
-      }
-      if (reserved1 != 0) {
-        throw new RuntimeException("Reserved1 field must be 0");
-      }
-      if (contentChecksum != 0) {
-        throw new RuntimeException("Content checksum is unsupported");
-      }
-      if (contentSize != 0) {
-        throw new RuntimeException("Content size is unsupported");
-      }
-      if (blockIndependence != 1) {
-        throw new RuntimeException("Dependent block stream is unsupported");
-      }
-      if (version != VERSION) {
-        throw new RuntimeException(String.format("Version %d is unsupported", version));
-      }
-    }
-    
-    public boolean isPresetDictionarySet() {
-      return presetDictionary == 1;
-    }
-    
-    public boolean isContentChecksumSet() {
-      return contentChecksum == 1;
-    }
-    
-    public boolean isContentSizeSet() {
-      return contentSize == 1;
-    }
-    
-    public boolean isBlockChecksumSet() {
-      return blockChecksum == 1;
-    }
-    
-    public boolean isBlockIndependenceSet() {
-      return blockIndependence == 1;
-    }
-    
-    public int getVersion() {
-      return version;
-    }
-  }
-  
-  public static class BD {
-    
-    private final int reserved2;
-    private final int blockSizeValue;
-    private final int reserved3;
-    
-    public BD() {
-      this(0, BLOCKSIZE_64KB, 0);
-    }
-    
-    public BD(int blockSizeValue) {
-      this(0, blockSizeValue, 0);
-    }
-    
-    private BD(int reserved2, int blockSizeValue, int reserved3) {
-      this.reserved2 = reserved2;
-      this.blockSizeValue = blockSizeValue;
-      this.reserved3 = reserved3;
-      validate();
-    }
-    
-    public static BD fromByte(byte bd) {
-      int reserved2 =        (bd >>> 0) & 15;
-      int blockMaximumSize = (bd >>> 4) & 7;
-      int reserved3 =        (bd >>> 7) & 1;
-      
-      return new BD(reserved2, blockMaximumSize, reserved3);
-    }
-    
-    private void validate() {
-      if (reserved2 != 0) {
-        throw new RuntimeException("Reserved2 field must be 0");
-      }
-      if (blockSizeValue < 4 || blockSizeValue > 7) {
-        throw new RuntimeException("Block size value must be between 4 and 7");
-      }
-      if (reserved3 != 0) {
-        throw new RuntimeException("Reserved3 field must be 0");
-      }
-    }
-    
-    // 2^(2n+8)
-    public int getBlockMaximumSize() {
-      return (1 << ((2 * blockSizeValue) + 8));
-    }
-    
-    public byte toByte() {
-      return (byte) (
-            ((reserved2       & 15) << 0)
-          | ((blockSizeValue  & 7) << 4)
-          | ((reserved3       & 1) << 7) );
-    }
-  }
-  
-}

http://git-wip-us.apache.org/repos/asf/kafka/blob/1c6d5bba/clients/src/main/java/org/apache/kafka/common/metrics/JmxReporter.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/metrics/JmxReporter.java b/clients/src/main/java/org/apache/kafka/common/metrics/JmxReporter.java
index 9c20538..6b9590c 100644
--- a/clients/src/main/java/org/apache/kafka/common/metrics/JmxReporter.java
+++ b/clients/src/main/java/org/apache/kafka/common/metrics/JmxReporter.java
@@ -42,7 +42,7 @@ import org.slf4j.LoggerFactory;
 public class JmxReporter implements MetricsReporter {
 
     private static final Logger log = LoggerFactory.getLogger(JmxReporter.class);
-    private static final Object lock = new Object();
+    private static final Object LOCK = new Object();
     private String prefix;
     private final Map<String, KafkaMbean> mbeans = new HashMap<String, KafkaMbean>();
 
@@ -58,12 +58,11 @@ public class JmxReporter implements MetricsReporter {
     }
 
     @Override
-    public void configure(Map<String, ?> configs) {
-    }
+    public void configure(Map<String, ?> configs) {}
 
     @Override
     public void init(List<KafkaMetric> metrics) {
-        synchronized (lock) {
+        synchronized (LOCK) {
             for (KafkaMetric metric : metrics)
                 addAttribute(metric);
             for (KafkaMbean mbean : mbeans.values())
@@ -73,7 +72,7 @@ public class JmxReporter implements MetricsReporter {
 
     @Override
     public void metricChange(KafkaMetric metric) {
-        synchronized (lock) {
+        synchronized (LOCK) {
             KafkaMbean mbean = addAttribute(metric);
             reregister(mbean);
         }
@@ -86,36 +85,35 @@ public class JmxReporter implements MetricsReporter {
             if (!this.mbeans.containsKey(mBeanName))
                 mbeans.put(mBeanName, new KafkaMbean(mBeanName));
             KafkaMbean mbean = this.mbeans.get(mBeanName);
-            mbean.setAttribute(metricName.name() , metric);
+            mbean.setAttribute(metricName.name(), metric);
             return mbean;
         } catch (JMException e) {
             throw new KafkaException("Error creating mbean attribute for metricName :" +
metric.metricName(), e);
         }
     }
 
-  /**
-   * @param metricName
-   * @return standard JMX MBean name in the following format
-   *       domainName:type=metricType,key1=val1,key2=val2
-   */
-  private String getMBeanName(MetricName metricName) {
-    StringBuilder mBeanName = new StringBuilder();
-    mBeanName.append(prefix);
-    mBeanName.append(":type=");
-    mBeanName.append(metricName.group());
-    for (Map.Entry<String, String> entry : metricName.tags().entrySet()) {
-      if(entry.getKey().length() <= 0 || entry.getValue().length() <= 0)
-         continue;
-      mBeanName.append(",");
-      mBeanName.append(entry.getKey());
-      mBeanName.append("=");
-      mBeanName.append(entry.getValue());
+    /**
+     * @param metricName
+     * @return standard JMX MBean name in the following format domainName:type=metricType,key1=val1,key2=val2
+     */
+    private String getMBeanName(MetricName metricName) {
+        StringBuilder mBeanName = new StringBuilder();
+        mBeanName.append(prefix);
+        mBeanName.append(":type=");
+        mBeanName.append(metricName.group());
+        for (Map.Entry<String, String> entry : metricName.tags().entrySet()) {
+            if (entry.getKey().length() <= 0 || entry.getValue().length() <= 0)
+                continue;
+            mBeanName.append(",");
+            mBeanName.append(entry.getKey());
+            mBeanName.append("=");
+            mBeanName.append(entry.getValue());
+        }
+        return mBeanName.toString();
     }
-    return mBeanName.toString();
-  }
 
     public void close() {
-        synchronized (lock) {
+        synchronized (LOCK) {
             for (KafkaMbean mbean : this.mbeans.values())
                 unregister(mbean);
         }
@@ -185,7 +183,12 @@ public class JmxReporter implements MetricsReporter {
             for (Map.Entry<String, KafkaMetric> entry : this.metrics.entrySet()) {
                 String attribute = entry.getKey();
                 KafkaMetric metric = entry.getValue();
-                attrs[i] = new MBeanAttributeInfo(attribute, double.class.getName(), metric.metricName().description(),
true, false, false);
+                attrs[i] = new MBeanAttributeInfo(attribute,
+                                                  double.class.getName(),
+                                                  metric.metricName().description(),
+                                                  true,
+                                                  false,
+                                                  false);
                 i += 1;
             }
             return new MBeanInfo(this.getClass().getName(), "", attrs, null, null, null);

http://git-wip-us.apache.org/repos/asf/kafka/blob/1c6d5bba/clients/src/main/java/org/apache/kafka/common/metrics/Sensor.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/metrics/Sensor.java b/clients/src/main/java/org/apache/kafka/common/metrics/Sensor.java
index e53cfaa..ca823fd 100644
--- a/clients/src/main/java/org/apache/kafka/common/metrics/Sensor.java
+++ b/clients/src/main/java/org/apache/kafka/common/metrics/Sensor.java
@@ -147,7 +147,7 @@ public final class Sensor {
      * @param stat The statistic to keep
      */
     public void add(MetricName metricName, MeasurableStat stat) {
-      add(metricName, stat, null);
+        add(metricName, stat, null);
     }
 
     /**
@@ -157,11 +157,11 @@ public final class Sensor {
      * @param config A special configuration for this metric. If null use the sensor default
configuration.
      */
     public synchronized void add(MetricName metricName, MeasurableStat stat, MetricConfig
config) {
-      KafkaMetric metric = new KafkaMetric(new Object(),
-                                      Utils.notNull(metricName),
-                                      Utils.notNull(stat),
-                                      config == null ? this.config : config,
-                                      time);
+        KafkaMetric metric = new KafkaMetric(new Object(),
+                                             Utils.notNull(metricName),
+                                             Utils.notNull(stat),
+                                             config == null ? this.config : config,
+                                             time);
         this.registry.registerMetric(metric);
         this.metrics.add(metric);
         this.stats.add(stat);

http://git-wip-us.apache.org/repos/asf/kafka/blob/1c6d5bba/clients/src/main/java/org/apache/kafka/common/metrics/stats/Rate.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/metrics/stats/Rate.java b/clients/src/main/java/org/apache/kafka/common/metrics/stats/Rate.java
index a5838b3..98429da 100644
--- a/clients/src/main/java/org/apache/kafka/common/metrics/stats/Rate.java
+++ b/clients/src/main/java/org/apache/kafka/common/metrics/stats/Rate.java
@@ -71,7 +71,7 @@ public class Rate implements MeasurableStat {
             case MILLISECONDS:
                 return time;
             case SECONDS:
-                return time / (1000.0);
+                return time / 1000.0;
             case MINUTES:
                 return time / (60.0 * 1000.0);
             case HOURS:

http://git-wip-us.apache.org/repos/asf/kafka/blob/1c6d5bba/clients/src/main/java/org/apache/kafka/common/network/NetworkReceive.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/network/NetworkReceive.java b/clients/src/main/java/org/apache/kafka/common/network/NetworkReceive.java
index dcc639a..fc0d168 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/NetworkReceive.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/NetworkReceive.java
@@ -50,7 +50,7 @@ public class NetworkReceive implements Receive {
 
     @Override
     public ByteBuffer[] reify() {
-        return new ByteBuffer[] { this.buffer };
+        return new ByteBuffer[] {this.buffer};
     }
 
     @Override

http://git-wip-us.apache.org/repos/asf/kafka/blob/1c6d5bba/clients/src/main/java/org/apache/kafka/common/network/Selector.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/network/Selector.java b/clients/src/main/java/org/apache/kafka/common/network/Selector.java
index e18a769..6baad93 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/Selector.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/Selector.java
@@ -14,7 +14,6 @@ package org.apache.kafka.common.network;
 
 import java.io.EOFException;
 import java.io.IOException;
-import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.Socket;
 import java.nio.channels.CancelledKeyException;
@@ -275,7 +274,7 @@ public class Selector implements Selectable {
                     }
                 } catch (IOException e) {
                     String desc = socketDescription(channel);
-                    if(e instanceof EOFException)
+                    if (e instanceof EOFException)
                         log.info("Connection {} disconnected", desc);
                     else
                         log.warn("Error in I/O with connection to {}", desc, e);
@@ -290,9 +289,9 @@ public class Selector implements Selectable {
     
     private String socketDescription(SocketChannel channel) {
         Socket socket = channel.socket();
-        if(socket == null)
+        if (socket == null)
             return "[unconnected socket]";
-        else if(socket.getInetAddress() != null)
+        else if (socket.getInetAddress() != null)
             return socket.getInetAddress().toString();
         else
             return socket.getLocalAddress().toString();
@@ -525,7 +524,7 @@ public class Selector implements Selectable {
                     String metricGrpName = metricGrpPrefix + "-node-metrics";
 
                     Map<String, String> tags = new LinkedHashMap<String, String>(metricTags);
-                    tags.put("node-id", "node-"+node);
+                    tags.put("node-id", "node-" + node);
 
                     nodeRequest = this.metrics.sensor(nodeRequestName);
                     MetricName metricName = new MetricName("outgoing-byte-rate", metricGrpName,
tags);

http://git-wip-us.apache.org/repos/asf/kafka/blob/1c6d5bba/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java b/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java
index 109fc96..07aba71 100644
--- a/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java
+++ b/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java
@@ -16,10 +16,6 @@
  */
 package org.apache.kafka.common.protocol;
 
-
-import java.util.ArrayList;
-import java.util.List;
-
 /**
  * Identifiers for all the Kafka APIs
  */
@@ -37,16 +33,18 @@ public enum ApiKeys {
     HEARTBEAT(12, "heartbeat");
 
     private static ApiKeys[] codeToType;
-    public static int MAX_API_KEY = -1;
+    public static final int MAX_API_KEY;
 
     static {
+        int maxKey = -1;
         for (ApiKeys key : ApiKeys.values()) {
-            MAX_API_KEY = Math.max(MAX_API_KEY, key.id);
+            maxKey = Math.max(maxKey, key.id);
         }
-        codeToType = new ApiKeys[MAX_API_KEY+1];
+        codeToType = new ApiKeys[maxKey + 1];
         for (ApiKeys key : ApiKeys.values()) {
             codeToType[key.id] = key;
         }
+        MAX_API_KEY = maxKey;
     }
 
     /** the perminant and immutable id of an API--this can't change ever */


Mime
View raw message