Merge pull request #1922 from emeroad/TraceV2/TraceFormatV2

#1819 trace format v2

commons-server/src/main/java/com/navercorp/pinpoint/common/server/bo/serializer/trace/v2/SpanDecoder.java

@@ -15,5 +15,4 @@ public interface SpanDecoder {
 
     void next(SpanDecodingContext decodingContext);
 
-    void finish(SpanDecodingContext decodingContext);
 }

commons-server/src/main/java/com/navercorp/pinpoint/common/server/bo/serializer/trace/v2/SpanDecoderV0.java

@@ -426,11 +426,6 @@ public void next(SpanDecodingContext decodingContext) {
         decodingContext.next();
     }
 
-    @Override
-    public void finish(SpanDecodingContext decodingContext) {
-        decodingContext.finish();
-    }
-
 
     // resolve type miss match
     private interface SpanAdaptor {

commons/src/main/java/com/navercorp/pinpoint/common/buffer/Buffer.java

@@ -194,6 +194,8 @@ public interface Buffer {
     @Deprecated
     void put(byte[] v);
 
+    byte getByte(int index);
+
     byte readByte();
 
     int readUnsignedByte();

commons/src/main/java/com/navercorp/pinpoint/common/buffer/FixedBuffer.java

@@ -303,6 +303,12 @@ public void put(final byte[] v) {
         putBytes(v);
     }
 
+
+    @Override
+    public byte getByte(int index) {
+        return this.buffer[offset];
+    }
+
     @Override
     public byte readByte() {
         return this.buffer[offset++];

web/src/main/java/com/navercorp/pinpoint/web/mapper/SpanMapperV2.java

@@ -80,6 +80,7 @@ public List<SpanBo> mapRow(Result result, int rowNum) throws Exception {
         decodingContext.setTransactionId(transactionId);
 
         for (Cell cell : rawCells) {
+            SpanDecoder spanDecoder = null;
             // only if family name is "span"
             if (CellUtil.matchingFamily(cell, HBaseTables.TRACE_V2_CF_SPAN)) {
 
@@ -88,20 +89,38 @@ public List<SpanBo> mapRow(Result result, int rowNum) throws Exception {
                 final Buffer qualifier = new OffsetFixedBuffer(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
                 final Buffer columnValue = new OffsetFixedBuffer(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
 
-                this.spanDecoder.decode(qualifier, columnValue, decodingContext, out);
+                spanDecoder = resolveDecoder(columnValue);
+                spanDecoder.decode(qualifier, columnValue, decodingContext, out);
 
             } else {
                 logger.warn("Unknown ColumnFamily :{}", Bytes.toStringBinary(CellUtil.cloneFamily(cell)));
             }
-            this.spanDecoder.next(decodingContext);
+            nextCell(spanDecoder, decodingContext);
         }
-        this.spanDecoder.finish(decodingContext);
+        decodingContext.finish();
 
 
         return buildSpanBoList(out);
 
     }
 
+    private void nextCell(SpanDecoder spanDecoder, SpanDecodingContext decodingContext) {
+        if (spanDecoder != null) {
+            spanDecoder.next(decodingContext);
+        } else {
+            decodingContext.next();
+        }
+    }
+
+    private SpanDecoder resolveDecoder(Buffer columnValue) {
+        final byte version = columnValue.getByte(0);
+        if (version == 0) {
+            return this.spanDecoder;
+        } else {
+            throw new IllegalStateException("unsupported version");
+        }
+    }
+
     private TransactionId newTransactionId(byte[] rowKey, int offset) {
 
         String agentId = BytesUtils.toStringAndRightTrim(rowKey, offset, AGENT_NAME_MAX_LEN);