Skip to content

MINOR: Fix ProcessorContext JavaDocs #8603

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mjsax merged 2 commits into from
May 12, 2020
Merged

MINOR: Fix ProcessorContext JavaDocs #8603

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7b80c61
MINOR: Fix ProcessorContext JavaDocs
mjsax May 2, 2020
60eabac
Github comments
mjsax May 6, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 24 additions & 20 deletions streams/src/main/java/org/apache/kafka/streams/processor/ProcessorContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,28 +34,28 @@
public interface ProcessorContext<K, V> {

/**
* Returns the application id
* Returns the application id.
*
* @return the application id
*/
String applicationId();

/**
* Returns the task id
* Returns the task id.
*
* @return the task id
*/
TaskId taskId();

/**
* Returns the default key serde
* Returns the default key serde.
*
* @return the key serializer
*/
Serde<?> keySerde();

/**
* Returns the default value serde
* Returns the default value serde.
*
* @return the value serializer
*/
Expand All @@ -69,7 +69,7 @@ public interface ProcessorContext<K, V> {
File stateDir();

/**
* Returns Metrics instance
* Returns Metrics instance.
*
* @return StreamsMetrics
*/
Expand Down Expand Up @@ -188,7 +188,8 @@ Cancellable schedule(final Duration interval,
<K1 extends K, V1 extends V> void forward(final K1 key, final V1 value, final To to);

/**
* Forwards a key/value pair to one of the downstream processors designated by childIndex
* Forwards a key/value pair to one of the downstream processors designated by childIndex.
*
* @param key key
* @param value value
* @param childIndex index in list of children of this node
Expand All @@ -199,7 +200,8 @@ Cancellable schedule(final Duration interval,
<K1 extends K, V1 extends V> void forward(final K1 key, final V1 value, final int childIndex);

/**
* Forwards a key/value pair to one of the downstream processors designated by the downstream processor name
* Forwards a key/value pair to one of the downstream processors designated by the downstream processor name.
*
* @param key key
* @param value value
* @param childName name of downstream processor
Expand All @@ -209,49 +211,52 @@ Cancellable schedule(final Duration interval,
<K1 extends K, V1 extends V> void forward(final K1 key, final V1 value, final String childName);

/**
* Requests a commit
* Requests a commit.
*/
void commit();

/**
* Returns the topic name of the current input record; could be null if it is not
* available (for example, if this method is invoked from the punctuate call)
* available (for example, if this method is invoked from the punctuate call).
*
* @return the topic name
*/
String topic();

/**
* Returns the partition id of the current input record; could be -1 if it is not
* available (for example, if this method is invoked from the punctuate call)
* available (for example, if this method is invoked from the punctuate call).
*
* @return the partition id
*/
int partition();

/**
* Returns the offset of the current input record; could be -1 if it is not
* available (for example, if this method is invoked from the punctuate call)
* available (for example, if this method is invoked from the punctuate call).
*
* @return the offset
*/
long offset();

/**
* Returns the headers of the current input record; could be null if it is not available
* Returns the headers of the current input record; could be null if it is not
* available (for example, if this method is invoked from the punctuate call).
*
* @return the headers
*/
Headers headers();

/**
* Returns the current timestamp.
*
* If it is triggered while processing a record streamed from the source processor, timestamp is defined as the timestamp of the current input record; the timestamp is extracted from
* <p> If it is triggered while processing a record streamed from the source processor,
* timestamp is defined as the timestamp of the current input record; the timestamp is extracted from
* {@link org.apache.kafka.clients.consumer.ConsumerRecord ConsumerRecord} by {@link TimestampExtractor}.
*
* If it is triggered while processing a record generated not from the source processor (for example,
* <p> If it is triggered while processing a record generated not from the source processor (for example,
* if this method is invoked from the punctuate call), timestamp is defined as the current
* task's stream time, which is defined as the smallest among all its input stream partition timestamps.
* task's stream time, which is defined as the largest timestamp of any record processed by the task.
*
* @return the timestamp
*/
Expand All @@ -260,10 +265,10 @@ Cancellable schedule(final Duration interval,
/**
* Returns all the application config properties as key/value pairs.
*
* The config properties are defined in the {@link org.apache.kafka.streams.StreamsConfig}
* <p> The config properties are defined in the {@link org.apache.kafka.streams.StreamsConfig}
* object and associated to the ProcessorContext.
* <p>
* The type of the values is dependent on the {@link org.apache.kafka.common.config.ConfigDef.Type type} of the property
*
* <p> The type of the values is dependent on the {@link org.apache.kafka.common.config.ConfigDef.Type type} of the property
* (e.g. the value of {@link org.apache.kafka.streams.StreamsConfig#DEFAULT_KEY_SERDE_CLASS_CONFIG DEFAULT_KEY_SERDE_CLASS_CONFIG}
* will be of type {@link Class}, even if it was specified as a String to
* {@link org.apache.kafka.streams.StreamsConfig#StreamsConfig(Map) StreamsConfig(Map)}).
Expand All @@ -276,12 +281,11 @@ Cancellable schedule(final Duration interval,
* Returns all the application config properties with the given key prefix, as key/value pairs
* stripping the prefix.
*
* The config properties are defined in the {@link org.apache.kafka.streams.StreamsConfig}
* <p> The config properties are defined in the {@link org.apache.kafka.streams.StreamsConfig}
* object and associated to the ProcessorContext.
*
* @param prefix the properties prefix
* @return the key/values matching the given prefix from the StreamsConfig properties.
*
*/
Map<String, Object> appConfigsWithPrefix(final String prefix);

Expand Down
3 changes: 1 addition & 2 deletions streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordQueue.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -122,6 +122,7 @@ int addRawRecords(final Iterable<ConsumerRecord<byte[], byte[]>> rawRecords) {
public StampedRecord poll() {
final StampedRecord recordToReturn = headRecord;
headRecord = null;
partitionTime = Math.max(partitionTime, recordToReturn.timestamp);

updateHead();

Expand Down Expand Up @@ -201,8 +202,6 @@ private void updateHead() {
continue;
}
headRecord = new StampedRecord(deserialized, timestamp);

partitionTime = Math.max(partitionTime, timestamp);
}
}

Expand Down
128 changes: 60 additions & 68 deletions streams/src/test/java/org/apache/kafka/streams/processor/internals/PartitionGroupTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@

import static org.apache.kafka.common.utils.Utils.mkEntry;
import static org.apache.kafka.common.utils.Utils.mkMap;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.equalTo;
import static org.junit.Assert.assertEquals;
Expand Down Expand Up @@ -98,8 +99,8 @@ public void testTimeTracking() {
private void testFirstBatch() {
StampedRecord record;
final PartitionGroup.RecordInfo info = new PartitionGroup.RecordInfo();
assertThat(group.numBuffered(), is(0));

assertEquals(0, group.numBuffered());
// add three 3 records with timestamp 1, 3, 5 to partition-1
final List<ConsumerRecord<byte[], byte[]>> list1 = Arrays.asList(
new ConsumerRecord<>("topic", 1, 1L, recordKey, recordValue),
Expand All @@ -120,39 +121,36 @@ private void testFirstBatch() {
// st: -1 since no records was being processed yet

verifyBuffered(6, 3, 3);
assertEquals(1L, group.partitionTimestamp(partition1));
assertEquals(2L, group.partitionTimestamp(partition2));
assertEquals(1L, group.headRecordOffset(partition1).longValue());
assertEquals(2L, group.headRecordOffset(partition2).longValue());
assertEquals(-1L, group.streamTime());
assertEquals(0.0, metrics.metric(lastLatenessValue).metricValue());
assertThat(group.partitionTimestamp(partition1), is(RecordQueue.UNKNOWN));
assertThat(group.partitionTimestamp(partition2), is(RecordQueue.UNKNOWN));
assertThat(group.headRecordOffset(partition1), is(1L));
assertThat(group.headRecordOffset(partition2), is(2L));
assertThat(group.streamTime(), is(RecordQueue.UNKNOWN));
assertThat(metrics.metric(lastLatenessValue).metricValue(), is(0.0));

// get one record, now the time should be advanced
record = group.nextRecord(info);
// 1:[3, 5]
// 2:[2, 4, 6]
// st: 1
assertEquals(partition1, info.partition());
assertEquals(3L, group.partitionTimestamp(partition1));
assertEquals(2L, group.partitionTimestamp(partition2));
assertEquals(3L, group.headRecordOffset(partition1).longValue());
assertEquals(2L, group.headRecordOffset(partition2).longValue());
assertEquals(1L, group.streamTime());
assertThat(info.partition(), equalTo(partition1));
assertThat(group.partitionTimestamp(partition1), is(1L));
assertThat(group.partitionTimestamp(partition2), is(RecordQueue.UNKNOWN));
assertThat(group.headRecordOffset(partition1), is(3L));
assertThat(group.headRecordOffset(partition2), is(2L));
verifyTimes(record, 1L, 1L);
verifyBuffered(5, 2, 3);
assertEquals(0.0, metrics.metric(lastLatenessValue).metricValue());
assertThat(metrics.metric(lastLatenessValue).metricValue(), is(0.0));

// get one record, now the time should be advanced
record = group.nextRecord(info);
// 1:[3, 5]
// 2:[4, 6]
// st: 2
assertEquals(partition2, info.partition());
assertEquals(3L, group.partitionTimestamp(partition1));
assertEquals(4L, group.partitionTimestamp(partition2));
assertEquals(3L, group.headRecordOffset(partition1).longValue());
assertEquals(4L, group.headRecordOffset(partition2).longValue());
assertEquals(2L, group.streamTime());
assertThat(info.partition(), equalTo(partition2));
assertThat(group.partitionTimestamp(partition1), is(1L));
assertThat(group.partitionTimestamp(partition2), is(2L));
assertThat(group.headRecordOffset(partition1), is(3L));
assertThat(group.headRecordOffset(partition2), is(4L));
verifyTimes(record, 2L, 2L);
verifyBuffered(4, 2, 2);
assertEquals(0.0, metrics.metric(lastLatenessValue).metricValue());
Expand All @@ -172,102 +170,96 @@ private void testSecondBatch() {
// 2:[4, 6]
// st: 2 (just adding records shouldn't change it)
verifyBuffered(6, 4, 2);
assertEquals(3L, group.partitionTimestamp(partition1));
assertEquals(4L, group.partitionTimestamp(partition2));
assertEquals(3L, group.headRecordOffset(partition1).longValue());
assertEquals(4L, group.headRecordOffset(partition2).longValue());
assertEquals(2L, group.streamTime());
assertEquals(0.0, metrics.metric(lastLatenessValue).metricValue());
assertThat(group.partitionTimestamp(partition1), is(1L));
assertThat(group.partitionTimestamp(partition2), is(2L));
assertThat(group.headRecordOffset(partition1), is(3L));
assertThat(group.headRecordOffset(partition2), is(4L));
assertThat(group.streamTime(), is(2L));
assertThat(metrics.metric(lastLatenessValue).metricValue(), is(0.0));

// get one record, time should be advanced
record = group.nextRecord(info);
// 1:[5, 2, 4]
// 2:[4, 6]
// st: 3
assertEquals(partition1, info.partition());
assertEquals(5L, group.partitionTimestamp(partition1));
assertEquals(4L, group.partitionTimestamp(partition2));
assertEquals(5L, group.headRecordOffset(partition1).longValue());
assertEquals(4L, group.headRecordOffset(partition2).longValue());
assertEquals(3L, group.streamTime());
assertThat(info.partition(), equalTo(partition1));
assertThat(group.partitionTimestamp(partition1), is(3L));
assertThat(group.partitionTimestamp(partition2), is(2L));
assertThat(group.headRecordOffset(partition1), is(5L));
assertThat(group.headRecordOffset(partition2), is(4L));
verifyTimes(record, 3L, 3L);
verifyBuffered(5, 3, 2);
assertEquals(0.0, metrics.metric(lastLatenessValue).metricValue());
assertThat(metrics.metric(lastLatenessValue).metricValue(), is(0.0));

// get one record, time should be advanced
record = group.nextRecord(info);
// 1:[5, 2, 4]
// 2:[6]
// st: 4
assertEquals(partition2, info.partition());
assertEquals(5L, group.partitionTimestamp(partition1));
assertEquals(6L, group.partitionTimestamp(partition2));
assertEquals(5L, group.headRecordOffset(partition1).longValue());
assertEquals(6L, group.headRecordOffset(partition2).longValue());
assertEquals(4L, group.streamTime());
assertThat(info.partition(), equalTo(partition2));
assertThat(group.partitionTimestamp(partition1), is(3L));
assertThat(group.partitionTimestamp(partition2), is(4L));
assertThat(group.headRecordOffset(partition1), is(5L));
assertThat(group.headRecordOffset(partition2), is(6L));
verifyTimes(record, 4L, 4L);
verifyBuffered(4, 3, 1);
assertEquals(0.0, metrics.metric(lastLatenessValue).metricValue());
assertThat(metrics.metric(lastLatenessValue).metricValue(), is(0.0));

// get one more record, time should be advanced
record = group.nextRecord(info);
// 1:[2, 4]
// 2:[6]
// st: 5
assertEquals(partition1, info.partition());
assertEquals(5L, group.partitionTimestamp(partition1));
assertEquals(6L, group.partitionTimestamp(partition2));
assertEquals(2L, group.headRecordOffset(partition1).longValue());
assertEquals(6L, group.headRecordOffset(partition2).longValue());
assertEquals(5L, group.streamTime());
assertThat(info.partition(), equalTo(partition1));
assertThat(group.partitionTimestamp(partition1), is(5L));
assertThat(group.partitionTimestamp(partition2), is(4L));
assertThat(group.headRecordOffset(partition1), is(2L));
assertThat(group.headRecordOffset(partition2), is(6L));
verifyTimes(record, 5L, 5L);
verifyBuffered(3, 2, 1);
assertEquals(0.0, metrics.metric(lastLatenessValue).metricValue());
assertThat(metrics.metric(lastLatenessValue).metricValue(), is(0.0));

// get one more record, time should not be advanced
record = group.nextRecord(info);
// 1:[4]
// 2:[6]
// st: 5
assertEquals(partition1, info.partition());
assertEquals(5L, group.partitionTimestamp(partition1));
assertEquals(6L, group.partitionTimestamp(partition2));
assertEquals(4L, group.headRecordOffset(partition1).longValue());
assertEquals(6L, group.headRecordOffset(partition2).longValue());
assertEquals(5L, group.streamTime());
assertThat(info.partition(), equalTo(partition1));
assertThat(group.partitionTimestamp(partition1), is(5L));
assertThat(group.partitionTimestamp(partition2), is(4L));
assertThat(group.headRecordOffset(partition1), is(4L));
assertThat(group.headRecordOffset(partition2), is(6L));
verifyTimes(record, 2L, 5L);
verifyBuffered(2, 1, 1);
assertEquals(3.0, metrics.metric(lastLatenessValue).metricValue());
assertThat(metrics.metric(lastLatenessValue).metricValue(), is(3.0));

// get one more record, time should not be advanced
record = group.nextRecord(info);
// 1:[]
// 2:[6]
// st: 5
assertEquals(partition1, info.partition());
assertEquals(5L, group.partitionTimestamp(partition1));
assertEquals(6L, group.partitionTimestamp(partition2));
assertThat(info.partition(), equalTo(partition1));
assertThat(group.partitionTimestamp(partition1), is(5L));
assertThat(group.partitionTimestamp(partition2), is(4L));
assertNull(group.headRecordOffset(partition1));
assertEquals(6L, group.headRecordOffset(partition2).longValue());
assertEquals(5L, group.streamTime());
assertThat(group.headRecordOffset(partition2), is(6L));
verifyTimes(record, 4L, 5L);
verifyBuffered(1, 0, 1);
assertEquals(1.0, metrics.metric(lastLatenessValue).metricValue());
assertThat(metrics.metric(lastLatenessValue).metricValue(), is(1.0));

// get one more record, time should be advanced
record = group.nextRecord(info);
// 1:[]
// 2:[]
// st: 6
assertEquals(partition2, info.partition());
assertEquals(5L, group.partitionTimestamp(partition1));
assertEquals(6L, group.partitionTimestamp(partition2));
assertThat(info.partition(), equalTo(partition2));
assertThat(group.partitionTimestamp(partition1), is(5L));
assertThat(group.partitionTimestamp(partition2), is(6L));
assertNull(group.headRecordOffset(partition1));
assertNull(group.headRecordOffset(partition2));
assertEquals(6L, group.streamTime());
verifyTimes(record, 6L, 6L);
verifyBuffered(0, 0, 0);
assertEquals(0.0, metrics.metric(lastLatenessValue).metricValue());
assertThat(metrics.metric(lastLatenessValue).metricValue(), is(0.0));
}

@Test
Expand Down Expand Up @@ -319,8 +311,8 @@ record = group.nextRecord(info);
}

private void verifyTimes(final StampedRecord record, final long recordTime, final long streamTime) {
assertEquals(recordTime, record.timestamp);
assertEquals(streamTime, group.streamTime());
assertThat(record.timestamp, is(recordTime));
assertThat(group.streamTime(), is(streamTime));
}

private void verifyBuffered(final int totalBuffered, final int partitionOneBuffered, final int partitionTwoBuffered) {
Expand Down
Loading