[MINOR] Fix static code analysis warnings, part I

src/main/java/org/apache/sysds/hops/codegen/opt/ReachabilityGraph.java

@@ -304,8 +304,8 @@ private static class CutSet {
 		private CutSet(InterestingPoint[] cutPoints, 
 				InterestingPoint[] l, InterestingPoint[] r) {
 			cut = cutPoints;
-			left = (InterestingPoint[]) ArrayUtils.addAll(cut, l);
-			right = (InterestingPoint[]) ArrayUtils.addAll(cut, r);
+			left = ArrayUtils.addAll(cut, l);
+			right = ArrayUtils.addAll(cut, r);
 		}
 
 		private void updatePositions(HashMap<InterestingPoint,Integer> probe) {

src/main/java/org/apache/sysds/hops/recompile/Recompiler.java

@@ -130,16 +130,16 @@ public class Recompiler {
 	/** Local reused rewriter for dynamic rewrites during recompile */
 
 	/** Local DML configuration for thread-local config updates */
-	private static ThreadLocal<ProgramRewriter> _rewriter = new ThreadLocal<ProgramRewriter>() {
+	private static ThreadLocal<ProgramRewriter> _rewriter = new ThreadLocal<>() {
 		@Override protected ProgramRewriter initialValue() { return new ProgramRewriter(false, true); }
 	};
 
-	private static ThreadLocal<LopRewriter> _lopRewriter = new ThreadLocal<LopRewriter>() {
+	private static ThreadLocal<LopRewriter> _lopRewriter = new ThreadLocal<>() {
 		@Override protected LopRewriter initialValue() {return new LopRewriter();}
 	};
 
 	// additional reused objects to avoid repeated, incremental reallocation on deepCopyDags
-	private static ThreadLocal<HashMap<Long,Hop>> _memoHop = new ThreadLocal<HashMap<Long,Hop>>() {
+	private static ThreadLocal<HashMap<Long,Hop>> _memoHop = new ThreadLocal<>() {
 		@Override protected HashMap<Long,Hop> initialValue() { return new HashMap<>(); }
 		@Override public HashMap<Long,Hop> get() { var tmp = super.get(); tmp.clear(); return tmp; }
 	};

src/main/java/org/apache/sysds/hops/rewrite/HopRewriteUtils.java

@@ -569,8 +569,7 @@ public static DataOp createTransientRead(String name, MatrixBlock mb) {
 
 	public static DataOp createTransientRead(String name, MatrixObject mo) {
 		DataOp tread = new DataOp(name, DataType.MATRIX, ValueType.FP64, OpOpData.TRANSIENTREAD,
-			null, mo.getNumRows(), mo.getNumColumns(), mo.getNnz(), UpdateType.COPY,
-			(int)mo.getBlocksize());
+			null, mo.getNumRows(), mo.getNumColumns(), mo.getNnz(), UpdateType.COPY, mo.getBlocksize());
 		tread.setVisited();
 		copyLineNumbers(mo, tread);
 		tread.setFileName(name);

src/main/java/org/apache/sysds/hops/rewrite/RewriteElementwiseMultChainOptimization.java

@@ -253,7 +253,7 @@ private static Hop constructPower(final Hop hop, final int cnt) {
 	 */
 	//TODO replace by ComparableHop wrapper around hop that implements equals and compareTo
 	//in order to ensure comparisons that are 'consistent with equals'
-	private static final Comparator<Hop> compareByDataType = new Comparator<Hop>() {
+	private static final Comparator<Hop> compareByDataType = new Comparator<>() {
 		private final int[] orderDataType = new int[DataType.values().length];
 		{
 			for (int i = 0, valuesLength = DataType.values().length; i < valuesLength; i++)

src/main/java/org/apache/sysds/parser/dml/DmlSyntacticValidator.java

@@ -142,11 +142,11 @@ public class DmlSyntacticValidator implements DmlListener {
 	private static final String DEF_WORK_DIR = ".";
 
 	//externally loaded dml scripts filename (unmodified) / script
-	protected static ThreadLocal<HashMap<String, String>> _tScripts = new ThreadLocal<HashMap<String, String>>() {
+	protected static ThreadLocal<HashMap<String, String>> _tScripts = new ThreadLocal<>() {
 		@Override protected HashMap<String, String> initialValue() { return new HashMap<>(); }
 	};
 	//imported scripts to prevent infinite recursion, modified filename / namespace
-	protected static ThreadLocal<HashMap<String, String>> _f2NS = new ThreadLocal<HashMap<String, String>>() {
+	protected static ThreadLocal<HashMap<String, String>> _f2NS = new ThreadLocal<>() {
 		@Override protected HashMap<String, String> initialValue() { return new HashMap<>(); }
 	};
 

src/main/java/org/apache/sysds/runtime/codegen/LibSpoofPrimitives.java

@@ -47,7 +47,7 @@ public class LibSpoofPrimitives
 
 	//global pool of reusable vectors, individual operations set up their own thread-local
 	//ring buffers of reusable vectors with specific number of vectors and vector sizes 
-	private static ThreadLocal<VectorBuffer> memPool = new ThreadLocal<VectorBuffer>() {
+	private static ThreadLocal<VectorBuffer> memPool = new ThreadLocal<>() {
 		@Override protected VectorBuffer initialValue() { return new VectorBuffer(0,0,0); }
 	};
 

src/main/java/org/apache/sysds/runtime/compress/colgroup/APreAgg.java

@@ -173,7 +173,7 @@ private void tsmmAPreAgg(APreAgg lg, MatrixBlock result) {
 	private boolean shouldDirectMultiply(APreAgg lg, int nColL, int nColR, boolean leftPreAgg) {
 		int lMRows = lg.numRowsToMultiply();
 		int rMRows = this.numRowsToMultiply();
-		long commonDim = (long) Math.min(lMRows, rMRows);
+		long commonDim = Math.min(lMRows, rMRows);
 		long directFLOPS = commonDim * nColL * nColR * 2; // times 2 for first add then multiply
 
 		long preAggFLOPS = 0;
@@ -270,7 +270,7 @@ private void addMatrixToResult(MatrixBlock tmp, MatrixBlock result, IColIndex ro
 	private void tsmmColGroupUncompressed(ColGroupUncompressed other, MatrixBlock result) {
 		LOG.warn("Inefficient multiplication with uncompressed column group");
 		final int nCols = result.getNumColumns();
-		final MatrixBlock otherMBT = LibMatrixReorg.transpose(((ColGroupUncompressed) other).getData());
+		final MatrixBlock otherMBT = LibMatrixReorg.transpose(other.getData());
 		final int nRows = otherMBT.getNumRows();
 		final MatrixBlock tmp = new MatrixBlock(nRows, nCols, false);
 		tmp.allocateDenseBlock();

src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupDDCFOR.java

@@ -389,7 +389,7 @@ public AColGroup rexpandCols(int max, boolean ignore, boolean cast, int nRows) {
 				return ColGroupEmpty.create(max);
 			else {
 				double[] retDef = new double[max];
-				retDef[((int) def) - 1] = 1;
+				retDef[def - 1] = 1;
 				return ColGroupConst.create(retDef);
 			}
 		}
@@ -405,7 +405,7 @@ else if(def > max)
 				return ColGroupDDC.create(outCols, d, _data, getCachedCounts());
 			else {
 				double[] retDef = new double[max];
-				retDef[((int) def) - 1] = 1;
+				retDef[def - 1] = 1;
 				return ColGroupDDCFOR.create(outCols, d, _data, getCachedCounts(), retDef);
 			}
 		}

src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupRLE.java

@@ -350,8 +350,8 @@ protected final void computeRowSums(double[] c, int rl, int ru, double[] preAgg)
 				int re = tmp.astart;
 				for(; apos < blen; apos += 2) {
 					// for each run find new start and end
-					rs = re + (int) _data[apos];
-					re = rs + (int) _data[apos + 1];
+					rs = re + _data[apos];
+					re = rs + _data[apos + 1];
 					// TODO make specialized version that ignore rl if rl == 0.
 					// move start to new variable but minimum rl
 					final int rsc = Math.max(rs, rl); // runStartCorrected
@@ -739,17 +739,17 @@ public boolean containsValue(double pattern) {
 	}
 
 	private String pair(char[] d, int off) {
-		if((int) _data[off + 1] == 1)
+		if(_data[off + 1] == 1)
 			return ((int) _data[off]) + "";
 		else
 			return ((int) _data[off]) + "-" + ((int) _data[off + 1]);
 	}
 
 	private String pair(char[] d, int off, int sum) {
-		if((int) _data[off + 1] == 1)
-			return ((int) _data[off] + sum) + "";
+		if(_data[off + 1] == 1)
+			return (_data[off] + sum) + "";
 		else
-			return ((int) _data[off] + sum) + "-" + ((int) _data[off + 1]);
+			return (_data[off] + sum) + "-" + ((int) _data[off + 1]);
 	}
 
 	@Override

src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupSDC.java

@@ -496,7 +496,7 @@ protected static AColGroup rexpandCols(int max, boolean ignore, boolean cast, in
 				return ColGroupEmpty.create(max);
 			else {
 				double[] retDef = new double[max];
-				retDef[((int) def) - 1] = 1;
+				retDef[def - 1] = 1;
 				return ColGroupSDCSingle.create(ColIndexFactory.create(max), nRows, Dictionary.create(new double[max]),
 					retDef, indexes, null);
 			}
@@ -513,7 +513,7 @@ else if(def > max)
 				return ColGroupSDCZeros.create(outCols, nRows, d, indexes, data, counts);
 			else {
 				double[] retDef = new double[max];
-				retDef[((int) def) - 1] = 1;
+				retDef[def - 1] = 1;
 				return ColGroupSDC.create(outCols, nRows, d, retDef, indexes, data, counts);
 			}
 		}

src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupSDCFOR.java

@@ -101,7 +101,7 @@ else if(allZero)
 
 	public static AColGroup sparsifyFOR(ColGroupSDC g) {
 		// subtract default.
-		final double[] constV = ((ColGroupSDC) g)._defaultTuple;
+		final double[] constV = g._defaultTuple;
 		final AColGroupValue clg = (AColGroupValue) g.subtractDefaultTuple();
 		return create(g.getColIndices(), g._numRows, clg._dict, g._indexes, g._data, g.getCachedCounts(), constV);
 	}

src/main/java/org/apache/sysds/runtime/compress/colgroup/dictionary/IdentityDictionary.java

@@ -529,7 +529,7 @@ public ADictionary rexpandColsWithReference(int max, boolean ignore, boolean cas
 
 	@Override
 	public double getSparsity() {
-		return 1.0d / (double) nRowCol;
+		return 1d / nRowCol;
 	}
 
 	@Override

src/main/java/org/apache/sysds/runtime/compress/colgroup/dictionary/IdentityDictionarySlice.java

@@ -287,7 +287,7 @@ public ADictionary replaceWithReference(double pattern, double replace, double[]
 
 	@Override
 	public double getSparsity() {
-		return 1.0d / (double)nRowCol;
+		return 1d / nRowCol;
 	}
 
 	@Override

src/main/java/org/apache/sysds/runtime/compress/colgroup/dictionary/QDictionary.java

@@ -280,7 +280,7 @@ private double sumRowSq(int k, int nrColumns) {
 		int valOff = k * nrColumns;
 		double res = 0.0;
 		for(int i = 0; i < nrColumns; i++)
-			res += (int) (_values[valOff + i] * _values[valOff + i]) * _scale * _scale;
+			res += (_values[valOff + i] * _values[valOff + i]) * _scale * _scale;
 		return res;
 	}
 

src/main/java/org/apache/sysds/runtime/compress/colgroup/offset/AOffset.java

@@ -51,7 +51,7 @@ public abstract class AOffset implements Serializable {
 	protected static final Log LOG = LogFactory.getLog(AOffset.class.getName());
 
 	/** Thread local cache for a single recently used Iterator, this is used for cache blocking */
-	private ThreadLocal<OffsetCache> cacheRow = new ThreadLocal<OffsetCache>() {
+	private ThreadLocal<OffsetCache> cacheRow = new ThreadLocal<>() {
 		@Override
 		protected OffsetCache initialValue() {
 			return null;

src/main/java/org/apache/sysds/runtime/compress/colgroup/offset/OffsetChar.java

@@ -28,7 +28,7 @@
 public class OffsetChar extends AOffset {
 
 	private static final long serialVersionUID = -1192266421395964882L;
-	protected static final int maxV = (int) Character.MAX_VALUE;
+	protected static final int maxV = Character.MAX_VALUE;
 
 	private final char[] offsets;
 	private final int offsetToFirst;

src/main/java/org/apache/sysds/runtime/compress/estim/ComEstCompressed.java

@@ -40,7 +40,7 @@ protected ComEstCompressed(CompressedMatrixBlock data, CompressionSettings compS
 
 	@Override
 	protected List<CompressedSizeInfoColGroup> CompressedSizeInfoColGroup(int clen, int k) {
-		List<CompressedSizeInfoColGroup> ret = new ArrayList<CompressedSizeInfoColGroup>();
+		List<CompressedSizeInfoColGroup> ret = new ArrayList<>();
 		final int nRow = cData.getNumRows();
 		for(AColGroup g : cData.getColGroups()) {
 			ret.add(g.getCompressionInfo(nRow));

src/main/java/org/apache/sysds/runtime/compress/estim/ComEstFactory.java

@@ -64,7 +64,7 @@ public static AComEst createEstimator(MatrixBlock data, CompressionSettings cs,
 	}
 
 	private static AComEst createEstimator(MatrixBlock data, CompressionSettings cs, int sampleSize, int k, int nRows) {
-		if(sampleSize >= (double) nRows * 0.8) // if sample size is larger than 80% use entire input as sample.
+		if(sampleSize >= nRows * 0.8) // if sample size is larger than 80% use entire input as sample.
 			return createExactEstimator(data, cs);
 		else
 			return createSampleEstimator(data, cs, sampleSize, k);
@@ -132,7 +132,7 @@ private static int getSampleSize(double samplePower, int nRows, int nCols, doubl
 		sampleSize = (int) (sampleSize * (1.0 / Math.min(sparsity + 0.2, 1.0)));
 
 		// adhere to maximum sample size.
-		sampleSize = (int) Math.max(minSampleSize, Math.min(sampleSize, maxSampleSize));
+		sampleSize = Math.max(minSampleSize, Math.min(sampleSize, maxSampleSize));
 
 		return sampleSize;
 	}

src/main/java/org/apache/sysds/runtime/compress/estim/ComEstSample.java

@@ -353,7 +353,7 @@ private MatrixBlock denseSamplePath(int[] sampleRows) {
 		final double sampleRatio = _cs.transposed ? (double) _data.getNumColumns() /
 			sampleSize : (double) _data.getNumRows() / sampleSize;
 		final long inputNonZeros = _data.getNonZeros();
-		final long estimatedNonZerosInSample = (long) Math.ceil((double) inputNonZeros / sampleRatio);
+		final long estimatedNonZerosInSample = (long) Math.ceil(inputNonZeros / sampleRatio);
 		final int resRows = _cs.transposed ? _data.getNumRows() : _data.getNumColumns();
 		final long nCellsInSample = (long) sampleSize * resRows;
 		final boolean shouldBeSparseSample = 0.4 > (double) estimatedNonZerosInSample / nCellsInSample;

src/main/java/org/apache/sysds/runtime/compress/estim/CompressedSizeInfoColGroup.java

@@ -203,7 +203,7 @@ public boolean containsZeros() {
 
 	private static EnumMap<CompressionType, Double> calculateCompressionSizes(IColIndex cols, EstimationFactors fact,
 		Set<CompressionType> validCompressionTypes) {
-		EnumMap<CompressionType, Double> res = new EnumMap<CompressionType, Double>(CompressionType.class);
+		EnumMap<CompressionType, Double> res = new EnumMap<>(CompressionType.class);
 		for(CompressionType ct : validCompressionTypes) {
 			double compSize = getCompressionSize(cols, ct, fact);
 			if(compSize > 0)

src/main/java/org/apache/sysds/runtime/compress/estim/encoding/EncodingFactory.java

@@ -307,7 +307,7 @@ private static IEncode createFromSparse(MatrixBlock m, int col) {
 		final DoubleCountHashMap map = new DoubleCountHashMap(16);
 		final SparseBlock sb = m.getSparseBlock();
 
-		final double guessedNumberOfNonZero = Math.min(4, Math.ceil((double) m.getNumRows() * m.getSparsity()));
+		final double guessedNumberOfNonZero = Math.min(4, Math.ceil(m.getNumRows() * m.getSparsity()));
 		final IntArrayList offsets = new IntArrayList((int) guessedNumberOfNonZero);
 
 		// Iteration 1 of non zero values, make Count HashMap.

src/main/java/org/apache/sysds/runtime/compress/lib/CLALibAppend.java

@@ -63,7 +63,7 @@ else if(right.isEmpty() && left instanceof CompressedMatrixBlock)
 
 		// if compression failed then use default append method.
 		if(!(left instanceof CompressedMatrixBlock && right instanceof CompressedMatrixBlock)) {
-			final double spar = (double) (left.getNonZeros() + right.getNonZeros()) / ((double) m * n);
+			final double spar = (left.getNonZeros() + right.getNonZeros()) / ((double) m * n);
 			final double estSizeUncompressed = MatrixBlock.estimateSizeInMemory(m, n, spar);
 			final double estSizeCompressed = left.getInMemorySize() + right.getInMemorySize();
 			if(estSizeUncompressed < estSizeCompressed)

src/main/java/org/apache/sysds/runtime/compress/lib/CLALibCombineGroups.java

@@ -188,7 +188,7 @@ else if(ce instanceof ConstEncoding) {
 		else if(ce instanceof SparseEncoding) {
 			SparseEncoding sed = (SparseEncoding) ce;
 			ADictionary cd = DictionaryFactory.combineDictionariesSparse(ac, bc);
-			double[] defaultTuple = constructDefaultTuple((AColGroupCompressed) ac, (AColGroupCompressed) bc);
+			double[] defaultTuple = constructDefaultTuple(ac, bc);
 			return ColGroupSDC.create(combinedColumns, sed.getNumRows(), cd, defaultTuple, sed.getOffsets(), sed.getMap(),
 				null);
 		}

src/main/java/org/apache/sysds/runtime/compress/lib/CLALibDecompress.java

@@ -72,13 +72,8 @@ public static void decompressTo(CompressedMatrixBlock cmb, MatrixBlock ret, int
 			MatrixBlock mbSliced = cmb.slice( //
 				Math.min(Math.abs(rowOffset), 0), Math.min(cmb.getNumRows(), ret.getNumRows() - rowOffset) - 1, // Rows
 				Math.min(Math.abs(colOffset), 0), Math.min(cmb.getNumColumns(), ret.getNumColumns() - colOffset) - 1); // Cols
-			if(mbSliced instanceof MatrixBlock) {
-				mbSliced.putInto(ret, rowOffset, colOffset, false);
-				return;
-			}
-
-			cmb = (CompressedMatrixBlock) mbSliced;
-			decompress(cmb, 1);
+			mbSliced.putInto(ret, rowOffset, colOffset, false);
+			return;
 		}
 
 		final boolean outSparse = ret.isInSparseFormat();

src/main/java/org/apache/sysds/runtime/compress/lib/CLALibSlice.java

@@ -144,7 +144,7 @@ private static MatrixBlock sliceRowsDecompress(CompressedMatrixBlock cmb, int rl
 
 	private static MatrixBlock sliceRowsCompressed(CompressedMatrixBlock cmb, int rl, int ru) {
 		final List<AColGroup> groups = cmb.getColGroups();
-		final List<AColGroup> newColGroups = new ArrayList<AColGroup>(groups.size());
+		final List<AColGroup> newColGroups = new ArrayList<>(groups.size());
 		final int rue = ru + 1;
 
 		final CompressedMatrixBlock ret = new CompressedMatrixBlock(rue - rl, cmb.getNumColumns());

src/main/java/org/apache/sysds/runtime/controlprogram/caching/CacheableData.java

@@ -137,7 +137,7 @@ public enum CacheStatus {
 	/**
 	 * Current state of pinned variables, required for guarded collect.
 	 */
-	private static ThreadLocal<Long> sizePinned = new ThreadLocal<Long>() {
+	private static ThreadLocal<Long> sizePinned = new ThreadLocal<>() {
 		@Override protected Long initialValue() { return 0L; }
 	};
 

src/main/java/org/apache/sysds/runtime/controlprogram/caching/FrameObject.java

@@ -121,7 +121,7 @@ public ValueType[] getSchema(int cl, int cu) {
 	 * @return schema of value types
 	 */
 	public ValueType[] mergeSchemas(FrameObject fo) {
-		return (ValueType[]) ArrayUtils.addAll(
+		return ArrayUtils.addAll(
 			(_schema!=null) ? _schema : UtilFunctions.nCopies((int)getNumColumns(), ValueType.STRING), 
 			(fo._schema!=null) ? fo._schema : UtilFunctions.nCopies((int)fo.getNumColumns(), ValueType.STRING));
 	} 

src/main/java/org/apache/sysds/runtime/controlprogram/caching/UnifiedMemoryManager.java

@@ -97,7 +97,7 @@ public class UnifiedMemoryManager
 	// Operational memory limit in bytes (70%)
 	private static long _opMemLimit;
 	// List of pinned entries
-	private static final List<String> _pinnedEntries = new ArrayList<String>();
+	private static final List<String> _pinnedEntries = new ArrayList<>();
 
 	// Eviction queue of <filename,buffer> pairs (implemented via linked hash map
 	// for (1) queue semantics and (2) constant time get/insert/delete operations)

src/main/java/org/apache/sysds/runtime/controlprogram/context/ExecutionContext.java

@@ -365,7 +365,7 @@ public void setMetaData(String varName, long nrows, long ncols) {
 		MetaData oldMetaData = mo.getMetaData();
 		if( oldMetaData == null || !(oldMetaData instanceof MetaDataFormat) )
 			throw new DMLRuntimeException("Metadata not available");
-		MatrixCharacteristics mc = new MatrixCharacteristics(nrows, ncols, (int) mo.getBlocksize());
+		MatrixCharacteristics mc = new MatrixCharacteristics(nrows, ncols, mo.getBlocksize());
 		mo.setMetaData(new MetaDataFormat(mc, ((MetaDataFormat)oldMetaData).getFileFormat()));
 	}
 

src/main/java/org/apache/sysds/runtime/controlprogram/context/SparkExecutionContext.java

@@ -439,7 +439,7 @@ else if( mo.isDirty() || mo.isCached(false) || mo.isFederated() || mo instanceof
 			}
 			else { //default case
 				MatrixBlock mb = mo.acquireRead(); //pin matrix in memory
-				rdd = toMatrixJavaPairRDD(sc, mb, (int)mo.getBlocksize(), numParts, inclEmpty);
+				rdd = toMatrixJavaPairRDD(sc, mb, mo.getBlocksize(), numParts, inclEmpty);
 				mo.release(); //unpin matrix
 				_parRDDs.registerRDD(rdd.id(), OptimizerUtils.estimatePartitionedSizeExactSparsity(dc), true);
 			}
@@ -700,7 +700,7 @@ public PartitionedBroadcast<MatrixBlock> getBroadcastForMatrixObject(MatrixObjec
 					CacheableData.addBroadcastSize(-mo.getBroadcastHandle().getSize());
 
 				//obtain meta data for matrix
-				int blen = (int) mo.getBlocksize();
+				int blen = mo.getBlocksize();
 
 				//create partitioned matrix block and release memory consumed by input
 				MatrixBlock mb = mo.acquireRead();

src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedData.java

@@ -204,7 +204,7 @@ private static ChannelInitializer<SocketChannel> createChannel(InetSocketAddress
 		final int timeout = ConfigurationManager.getFederatedTimeout();
 		final boolean ssl = ConfigurationManager.isFederatedSSL();
 
-		return new ChannelInitializer<SocketChannel>() {
+		return new ChannelInitializer<>() {
 			@Override
 			protected void initChannel(SocketChannel ch) throws Exception {
 				final ChannelPipeline cp = ch.pipeline();

src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedWorker.java

@@ -184,7 +184,7 @@ private ChannelInitializer<SocketChannel> createChannel(boolean ssl) {
 			final SelfSignedCertificate cert = new SelfSignedCertificate();
 			final SslContext cont2 = SslContextBuilder.forServer(cert.certificate(), cert.privateKey()).build();
 
-			return new ChannelInitializer<SocketChannel>() {
+			return new ChannelInitializer<>() {
 				@Override
 				public void initChannel(SocketChannel ch) {
 					final ChannelPipeline cp = ch.pipeline();