fixed underflow issue

src/main/java/teetime/util/TcpReader.java

@@ -26,9 +26,11 @@ import java.nio.channels.SocketChannel;
 import kieker.common.exception.RecordInstantiationException;
 import kieker.common.logging.Log;
 import kieker.common.logging.LogFactory;
+import kieker.common.record.AbstractMonitoringRecord;
 import kieker.common.record.IMonitoringRecord;
 import kieker.common.record.factory.CachedRecordFactoryCatalog;
 import kieker.common.record.factory.IRecordFactory;
+import kieker.common.record.factory.old.RecordFactoryWrapper;
 import kieker.common.record.misc.RegistryRecord;
 import kieker.common.util.registry.ILookup;
 import kieker.common.util.registry.Lookup;
@@ -47,6 +49,9 @@ import teetime.stage.io.AbstractTcpReader;
 public class TcpReader extends AbstractTcpReader<IMonitoringRecord> {
 
 	private static final Logger LOGGER = LoggerFactory.getLogger(TcpReader.class);
+	private static final int INT_BYTES = AbstractMonitoringRecord.TYPE_SIZE_INT;
+	private static final int LONG_BYTES = AbstractMonitoringRecord.TYPE_SIZE_LONG;
+	private static final int UNKNOWN_SIZE = RecordFactoryWrapper.UNKNOWN_SIZE;
 
 	private final CachedRecordFactoryCatalog recordFactories = CachedRecordFactoryCatalog.getInstance();
 	// BETTER use a non thread-safe implementation to increase performance. A thread-safe version is not necessary.
@@ -87,18 +92,33 @@ public class TcpReader extends AbstractTcpReader<IMonitoringRecord> {
 
 	@Override
 	protected final boolean read(final ByteBuffer buffer) {
+		// identify record class
+		if (buffer.remaining() < INT_BYTES) {
+			return false;
+		}
 		final int clazzId = buffer.getInt();
+		final String recordClassName = this.stringRegistry.get(clazzId);
+
+		// identify logging timestamp
+		if (buffer.remaining() < LONG_BYTES) {
+			return false;
+		}
 		final long loggingTimestamp = buffer.getLong();
 
-		final String recordClassName = this.stringRegistry.get(clazzId);
+		// identify record data
+		final IRecordFactory<? extends IMonitoringRecord> recordFactory = this.recordFactories.get(recordClassName);
+		int recordSizeInBytes = recordFactory.getRecordSizeInBytes();
+		if (recordSizeInBytes != UNKNOWN_SIZE && buffer.remaining() < recordSizeInBytes) {
+			return false;
+		}
+
 		try {
-			final IRecordFactory<? extends IMonitoringRecord> recordFactory = this.recordFactories.get(recordClassName);
-			IMonitoringRecord record = recordFactory.create(buffer, this.stringRegistry);
+			final IMonitoringRecord record = recordFactory.create(buffer, this.stringRegistry);
 			record.setLoggingTimestamp(loggingTimestamp);
 
 			send(record);
 		} catch (final RecordInstantiationException ex) {
-			super.logger.error("Failed to create record:", ex);
+			super.logger.error("Failed to create: " + recordClassName, ex);
 		}
 
 		return true;