ColumnFilteringInterceptor implements Interceptor { @Override public Event intercept(Event event) { byte[] body = event.getBody(); if (body == null) { return event; } final int length = body.length; int i = 0; int index = 0; while (i < length) { if (body[i++] == srcSeparator) { positions[++index] = i; if (index >= maxIndex) { break; } } } int nextVirginIndex; boolean separatorTerminated; if (i == length && index < maxColumn) { nextVirginIndex = index + 2; positions[nextVirginIndex - 1] = length; separatorTerminated = length > 0 ? body[length - 1] != srcSeparator : false; } else { nextVirginIndex = index + 1; separatorTerminated = true; } int newArrayLen = 0; for (i = columns.length; i-- > 0;) { int column = columns[i]; int len = positions[column + 1] - positions[column]; if (len <= 0) { newArrayLen++; } else { if (separatorTerminated && positions[column + 1] == length) { newArrayLen++; } newArrayLen += len; } } byte[] newbody = new byte[newArrayLen]; int newoffset = 0; for (int column: columns) { int len = positions[column + 1] - positions[column]; if (len > 0) { System.arraycopy(body, positions[column], newbody, newoffset, len); newoffset += len; if (newbody[newoffset - 1] == srcSeparator) { newbody[newoffset - 1] = dstSeparator; } else { newbody[newoffset++] = dstSeparator; } } else { newbody[newoffset++] = dstSeparator; } } event.setBody(newbody); Arrays.fill(positions, 1, nextVirginIndex, 0); return event; } private  ColumnFilteringInterceptor(int[] columns, byte srcSeparator, byte dstSeparator); @Override void initialize(); @Override Event intercept(Event event); @Override List<Event> intercept(List<Event> events); @Override void close();  }
@Test public void testInterceptEventWithColumnZero() { HashMap<String, String> contextMap = new HashMap<String, String>(); contextMap.put(ColumnFilteringInterceptor.Constants.DST_SEPARATOR, Byte.toString((byte)1)); contextMap.put(ColumnFilteringInterceptor.Constants.SRC_SEPARATOR, Byte.toString((byte)2)); contextMap.put(ColumnFilteringInterceptor.Constants.COLUMNS, "0"); ColumnFilteringInterceptor.Builder builder = new ColumnFilteringInterceptor.Builder(); builder.configure(new Context(contextMap)); Interceptor interceptor = builder.build(); assertArrayEquals("Empty Bytes", "\001".getBytes(), interceptor.intercept(new InterceptorTestHelper.MyEvent("".getBytes())).getBody()); assertArrayEquals("One Field", "First\001".getBytes(), interceptor.intercept(new InterceptorTestHelper.MyEvent("First".getBytes())).getBody()); assertArrayEquals("Two Fields", "\001".getBytes(), interceptor.intercept(new InterceptorTestHelper.MyEvent("\002First".getBytes())).getBody()); }