@ Override public synchronized long commit ( long offset ) { if ( offset <= submittedOffset && offset > safeOffset ) { int ggap = ( int ) Math . min ( committedOffsets . length , offset - safeOffset ) ; if ( ggap == committedOffsets . length ) { LOGGER . rateLimitedLog ( LOG_SUPPRESSION_INTERVAL_SECONDS , Level . WARN , null , "CommitTracker moving topic commit point from %d to %d for topic " + topic + " partition " + partition + " group:" + consumerGroup , safeOffset , ( offset - committedOffsets . length + 1 ) ) ; safeOffset = offset - committedOffsets . length + 1 ; committedOffsets [ idx ( safeOffset ) ] = safeOffset ; } committedOffsets [ idx ( offset ) ] = offset ; while ( ggap > 0 && committedOffsets [ idx ( safeOffset ) ] + 1 == committedOffsets [ idx ( safeOffset + 1 ) ] ) { ++ safeOffset ; } if ( offerOffset >= 0 && ( offerOffset - safeOffset ) < committedOffsets . length ) { offerOffset = - 1L ; notify ( ) ; } } if ( offset == firstOffset ) { firstOffsetCommitted = true ; } return safeOffset ; }
VoltDB . It will be recorded in committedOffsets and calculate the offset - safeOffset which is safe to commit to Kafka