@ -23,7 +23,6 @@ import org.apache.kafka.common.requests.ProduceResponse.RecordError;
@@ -23,7 +23,6 @@ import org.apache.kafka.common.requests.ProduceResponse.RecordError;
import java.util.Collections ;
import java.util.List ;
import java.util.Optional ;
import java.util.OptionalInt ;
/ * *
@ -31,12 +30,11 @@ import java.util.OptionalInt;
@@ -31,12 +30,11 @@ import java.util.OptionalInt;
* /
public class LogAppendInfo {
public static final LogAppendInfo UNKNOWN_LOG_APPEND_INFO = new LogAppendInfo ( Optional . empty ( ) , - 1 , OptionalInt . empty ( ) ,
public static final LogAppendInfo UNKNOWN_LOG_APPEND_INFO = new LogAppendInfo ( - 1 , - 1 , OptionalInt . empty ( ) ,
RecordBatch . NO_TIMESTAMP , - 1L , RecordBatch . NO_TIMESTAMP , - 1L ,
RecordConversionStats . EMPTY , CompressionType . NONE , CompressionType . NONE , - 1 , - 1 ,
false , - 1L ) ;
RecordConversionStats . EMPTY , CompressionType . NONE , - 1 , - 1L ) ;
private Optional < LogOffsetMetadata > firstOffset ;
private long firstOffset ;
private long lastOffset ;
private long maxTimestamp ;
private long offsetOfMaxTimestamp ;
@ -46,21 +44,16 @@ public class LogAppendInfo {
@@ -46,21 +44,16 @@ public class LogAppendInfo {
private final OptionalInt lastLeaderEpoch ;
private final CompressionType sourceCompression ;
private final CompressionType targetCompression ;
private final int shallowCount ;
private final int validBytes ;
private final boolean offsetsMonotonic ;
private final long lastOffsetOfFirstBatch ;
private final List < RecordError > recordErrors ;
private final String errorMessage ;
private final LeaderHwChange leaderHwChange ;
/ * *
* Creates an instance with the given params .
*
* @param firstOffset The first offset in the message set unless the message format is less than V2 and we are appending
* to the follower . If the message is a duplicate message the segment base offset and relative position
* in segment will be unknown .
* to the follower .
* @param lastOffset The last offset in the message set
* @param lastLeaderEpoch The partition leader epoch corresponding to the last offset , if available .
* @param maxTimestamp The maximum timestamp of the message set .
@ -69,13 +62,10 @@ public class LogAppendInfo {
@@ -69,13 +62,10 @@ public class LogAppendInfo {
* @param logStartOffset The start offset of the log at the time of this append .
* @param recordConversionStats Statistics collected during record processing , ` null ` if ` assignOffsets ` is ` false `
* @param sourceCompression The source codec used in the message set ( send by the producer )
* @param targetCompression The target codec of the message set ( after applying the broker compression configuration if any )
* @param shallowCount The number of shallow messages
* @param validBytes The number of valid bytes
* @param offsetsMonotonic Are the offsets in this message set monotonically increasing
* @param lastOffsetOfFirstBatch The last offset of the first batch
* /
public LogAppendInfo ( Optional < LogOffsetMetadata > firstOffset ,
public LogAppendInfo ( long firstOffset ,
long lastOffset ,
OptionalInt lastLeaderEpoch ,
long maxTimestamp ,
@ -84,22 +74,18 @@ public class LogAppendInfo {
@@ -84,22 +74,18 @@ public class LogAppendInfo {
long logStartOffset ,
RecordConversionStats recordConversionStats ,
CompressionType sourceCompression ,
CompressionType targetCompression ,
int shallowCount ,
int validBytes ,
boolean offsetsMonotonic ,
long lastOffsetOfFirstBatch ) {
this ( firstOffset , lastOffset , lastLeaderEpoch , maxTimestamp , offsetOfMaxTimestamp , logAppendTime , logStartOffset ,
recordConversionStats , sourceCompression , targetCompression , shallowCount , validBytes , offsetsMonotonic ,
lastOffsetOfFirstBatch , Collections . < RecordError > emptyList ( ) , null , LeaderHwChange . NONE ) ;
recordConversionStats , sourceCompression , validBytes , lastOffsetOfFirstBatch , Collections . < RecordError > emptyList ( ) ,
LeaderHwChange . NONE ) ;
}
/ * *
* Creates an instance with the given params .
*
* @param firstOffset The first offset in the message set unless the message format is less than V2 and we are appending
* to the follower . If the message is a duplicate message the segment base offset and relative position
* in segment will be unknown .
* to the follower .
* @param lastOffset The last offset in the message set
* @param lastLeaderEpoch The partition leader epoch corresponding to the last offset , if available .
* @param maxTimestamp The maximum timestamp of the message set .
@ -108,17 +94,13 @@ public class LogAppendInfo {
@@ -108,17 +94,13 @@ public class LogAppendInfo {
* @param logStartOffset The start offset of the log at the time of this append .
* @param recordConversionStats Statistics collected during record processing , ` null ` if ` assignOffsets ` is ` false `
* @param sourceCompression The source codec used in the message set ( send by the producer )
* @param targetCompression The target codec of the message set ( after applying the broker compression configuration if any )
* @param shallowCount The number of shallow messages
* @param validBytes The number of valid bytes
* @param offsetsMonotonic Are the offsets in this message set monotonically increasing
* @param lastOffsetOfFirstBatch The last offset of the first batch
* @param errorMessage error message
* @param recordErrors List of record errors that caused the respective batch to be dropped
* @param leaderHwChange Incremental if the high watermark needs to be increased after appending record
* Same if high watermark is not changed . None is the default value and it means append failed
* /
public LogAppendInfo ( Optional < LogOffsetMetadata > firstOffset ,
public LogAppendInfo ( long firstOffset ,
long lastOffset ,
OptionalInt lastLeaderEpoch ,
long maxTimestamp ,
@ -127,13 +109,9 @@ public class LogAppendInfo {
@@ -127,13 +109,9 @@ public class LogAppendInfo {
long logStartOffset ,
RecordConversionStats recordConversionStats ,
CompressionType sourceCompression ,
CompressionType targetCompression ,
int shallowCount ,
int validBytes ,
boolean offsetsMonotonic ,
long lastOffsetOfFirstBatch ,
List < RecordError > recordErrors ,
String errorMessage ,
LeaderHwChange leaderHwChange ) {
this . firstOffset = firstOffset ;
this . lastOffset = lastOffset ;
@ -144,21 +122,17 @@ public class LogAppendInfo {
@@ -144,21 +122,17 @@ public class LogAppendInfo {
this . logStartOffset = logStartOffset ;
this . recordConversionStats = recordConversionStats ;
this . sourceCompression = sourceCompression ;
this . targetCompression = targetCompression ;
this . shallowCount = shallowCount ;
this . validBytes = validBytes ;
this . offsetsMonotonic = offsetsMonotonic ;
this . lastOffsetOfFirstBatch = lastOffsetOfFirstBatch ;
this . recordErrors = recordErrors ;
this . errorMessage = errorMessage ;
this . leaderHwChange = leaderHwChange ;
}
public Optional < LogOffsetMetadata > firstOffset ( ) {
public long firstOffset ( ) {
return firstOffset ;
}
public void setFirstOffset ( Optional < LogOffsetMetadata > firstOffset ) {
public void setFirstOffset ( long firstOffset ) {
this . firstOffset = firstOffset ;
}
@ -218,30 +192,14 @@ public class LogAppendInfo {
@@ -218,30 +192,14 @@ public class LogAppendInfo {
return sourceCompression ;
}
public CompressionType targetCompression ( ) {
return targetCompression ;
}
public int shallowCount ( ) {
return shallowCount ;
}
public int validBytes ( ) {
return validBytes ;
}
public boolean offsetsMonotonic ( ) {
return offsetsMonotonic ;
}
public List < RecordError > recordErrors ( ) {
return recordErrors ;
}
public String errorMessage ( ) {
return errorMessage ;
}
public LeaderHwChange leaderHwChange ( ) {
return leaderHwChange ;
}
@ -253,7 +211,7 @@ public class LogAppendInfo {
@@ -253,7 +211,7 @@ public class LogAppendInfo {
* offset to avoid decompressing the data .
* /
public long firstOrLastOffsetOfFirstBatch ( ) {
return firstOffset . map ( x - > x . messageOffset ) . orElse ( lastOffsetOfFirstBatch ) ;
return firstOffset > = 0 ? firstOffset : lastOffsetOfFirstBatch ;
}
/ * *
@ -262,8 +220,8 @@ public class LogAppendInfo {
@@ -262,8 +220,8 @@ public class LogAppendInfo {
* @return Maximum possible number of messages described by LogAppendInfo
* /
public long numMessages ( ) {
if ( firstOffset . isPresent ( ) & & firstOffset . get ( ) . messageOffset > = 0 & & lastOffset > = 0 ) {
return lastOffset - firstOffset . get ( ) . messageOffset + 1 ;
if ( firstOffset > = 0 & & lastOffset > = 0 ) {
return lastOffset - firstOffset + 1 ;
}
return 0 ;
}
@ -276,24 +234,22 @@ public class LogAppendInfo {
@@ -276,24 +234,22 @@ public class LogAppendInfo {
* /
public LogAppendInfo copy ( LeaderHwChange newLeaderHwChange ) {
return new LogAppendInfo ( firstOffset , lastOffset , lastLeaderEpoch , maxTimestamp , offsetOfMaxTimestamp , logAppendTime , logStartOffset , recordConversionStats ,
sourceCompression , targetCompression , shallowCount , validBytes , offsetsMonotonic , lastOffsetOfFirstBatch , recordErrors , errorMessage , newLeaderHwChange ) ;
sourceCompression , validBytes , lastOffsetOfFirstBatch , recordErrors , newLeaderHwChange ) ;
}
public static LogAppendInfo unknownLogAppendInfoWithLogStartOffset ( long logStartOffset ) {
return new LogAppendInfo ( Optional . empty ( ) , - 1 , OptionalInt . empty ( ) , RecordBatch . NO_TIMESTAMP , - 1L , RecordBatch . NO_TIMESTAMP , logStartOffset ,
RecordConversionStats . EMPTY , CompressionType . NONE , CompressionType . NONE , - 1 , - 1 ,
false , - 1L ) ;
return new LogAppendInfo ( - 1 , - 1 , OptionalInt . empty ( ) , RecordBatch . NO_TIMESTAMP , - 1L , RecordBatch . NO_TIMESTAMP , logStartOffset ,
RecordConversionStats . EMPTY , CompressionType . NONE , - 1 , - 1L ) ;
}
/ * *
* In ProduceResponse V8 + , we add two new fields record_errors and error_message ( see KIP - 467 ) .
* For any record failures with InvalidTimestamp or InvalidRecordException , we construct a LogAppendInfo object like the one
* in unknownLogAppendInfoWithLogStartOffset , but with additional fields recordErrors and errorMessage
* in unknownLogAppendInfoWithLogStartOffset , but with additional fields recordErrors
* /
public static LogAppendInfo unknownLogAppendInfoWithAdditionalInfo ( long logStartOffset , List < RecordError > recordErrors , String errorMessage ) {
return new LogAppendInfo ( Optional . empty ( ) , - 1 , OptionalInt . empty ( ) , RecordBatch . NO_TIMESTAMP , - 1L , RecordBatch . NO_TIMESTAMP , logStartOffset ,
RecordConversionStats . EMPTY , CompressionType . NONE , CompressionType . NONE , - 1 , - 1 ,
false , - 1L , recordErrors , errorMessage , LeaderHwChange . NONE ) ;
public static LogAppendInfo unknownLogAppendInfoWithAdditionalInfo ( long logStartOffset , List < RecordError > recordErrors ) {
return new LogAppendInfo ( - 1 , - 1 , OptionalInt . empty ( ) , RecordBatch . NO_TIMESTAMP , - 1L , RecordBatch . NO_TIMESTAMP , logStartOffset ,
RecordConversionStats . EMPTY , CompressionType . NONE , - 1 , - 1L , recordErrors , LeaderHwChange . NONE ) ;
}
@Override
@ -308,13 +264,9 @@ public class LogAppendInfo {
@@ -308,13 +264,9 @@ public class LogAppendInfo {
", logStartOffset=" + logStartOffset +
", recordConversionStats=" + recordConversionStats +
", sourceCompression=" + sourceCompression +
", targetCompression=" + targetCompression +
", shallowCount=" + shallowCount +
", validBytes=" + validBytes +
", offsetsMonotonic=" + offsetsMonotonic +
", lastOffsetOfFirstBatch=" + lastOffsetOfFirstBatch +
", recordErrors=" + recordErrors +
", errorMessage='" + errorMessage + '\'' +
", leaderHwChange=" + leaderHwChange +
')' ;
}
David Mao
1 year ago
committed by
GitHub