root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 3 Packages Projects Releases Wiki Activity

MINOR: avoid unnecessary UnsupportedOperationException (#15102) 

We did no complete KIP-714 with regard to collecting producer clients
instance IDs in Kafka Streams if EOSv1 is enabled. Instead of throwing
an UnsupportedOperationException, we should return an empty map.

Reviewers: Lucas Brutschy <lbrutschy@confluent.io>
This commit is contained in:
Matthias J. Sax 2024-01-02 07:34:42 -08:00 committed by GitHub
parent 735022a423
commit 31ab73cf1e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 34 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
View File

@ -1877,6 +1877,9 @@ public class KafkaStreams implements AutoCloseable {
/** /**
* Returns the internal clients' assigned {@code client instance ids}. * Returns the internal clients' assigned {@code client instance ids}.
* <p>
* Note, if {@link StreamsConfig#PROCESSING_GUARANTEE_CONFIG} is set to {@link StreamsConfig#EXACTLY_ONCE},
* the producer client instance ids are not returned yet. This gap will be closed in the next release.
* *
* @return The internal clients' assigned instance ids used for metrics collection. * @return The internal clients' assigned instance ids used for metrics collection.
* *

4
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
View File

@ -1604,7 +1604,9 @@ public class StreamThread extends Thread implements ProcessingThread {
} }
if (processingMode.equals(StreamsConfigUtils.ProcessingMode.EXACTLY_ONCE_ALPHA)) { if (processingMode.equals(StreamsConfigUtils.ProcessingMode.EXACTLY_ONCE_ALPHA)) {
throw new UnsupportedOperationException("not yet implemented"); log.error("EOS v1 enabled: Producer client instance IDs are not collected." +
" Enable producer logging to retrieve the IDs from the producer logs.");
producerInstanceIdFuture.complete(Collections.emptyMap());
} else { } else {
if (threadProducerInstanceIdFuture.isDone()) { if (threadProducerInstanceIdFuture.isDone()) {
if (threadProducerInstanceIdFuture.isCompletedExceptionally()) { if (threadProducerInstanceIdFuture.isCompletedExceptionally()) {

39
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
View File

@ -3340,24 +3340,37 @@ public class StreamThreadTest {
@Test @Test
public void shouldGetProducerInstanceId() throws Exception { public void shouldGetProducerInstanceId() throws Exception {
getProducerInstanceId(false); getProducerInstanceId(false, false);
} }
@Test @Test
public void shouldProducerInstanceIdAndInternalTimeout() throws Exception { public void shouldGetProducerInstanceIdWithInternalTimeout() throws Exception {
getProducerInstanceId(true); getProducerInstanceId(true, false);
} }
private void getProducerInstanceId(final boolean injectTimeException) throws Exception { @Test
public void shouldNotGetProducerInstanceIdWithEosV1() throws Exception {
getProducerInstanceId(false, true);
}
@SuppressWarnings("deprecation")
private void getProducerInstanceId(final boolean injectTimeException,
final boolean enableEos) throws Exception {
final Uuid producerInstanceId = Uuid.randomUuid(); final Uuid producerInstanceId = Uuid.randomUuid();
final MockProducer<byte[], byte[]> producer = new MockProducer<>(); final MockProducer<byte[], byte[]> producer = new MockProducer<>();
producer.setClientInstanceId(producerInstanceId); if (!enableEos) {
if (injectTimeException) { producer.setClientInstanceId(producerInstanceId);
producer.injectTimeoutException(1); if (injectTimeException) {
producer.injectTimeoutException(1);
}
} }
clientSupplier.prepareProducer(producer); clientSupplier.prepareProducer(producer);
thread = createStreamThread("clientId"); final Properties properties = configProps(enableEos);
if (enableEos) {
properties.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE);
}
thread = createStreamThread("clientId", new StreamsConfig(properties));
thread.setState(State.STARTING); thread.setState(State.STARTING);
final KafkaFuture<Map<String, KafkaFuture<Uuid>>> producerInstanceIdFutures = final KafkaFuture<Map<String, KafkaFuture<Uuid>>> producerInstanceIdFutures =
@ -3366,9 +3379,13 @@ public class StreamThreadTest {
thread.maybeGetClientInstanceIds(); // triggers internal timeout; should not crash thread.maybeGetClientInstanceIds(); // triggers internal timeout; should not crash
thread.maybeGetClientInstanceIds(); thread.maybeGetClientInstanceIds();
final KafkaFuture<Uuid> producerFuture = producerInstanceIdFutures.get().get("clientId-StreamThread-1-producer"); if (enableEos) {
final Uuid producerUuid = producerFuture.get(); assertThat(producerInstanceIdFutures.get(), equalTo(emptyMap()));
assertThat(producerUuid, equalTo(producerInstanceId)); } else {
final KafkaFuture<Uuid> producerFuture = producerInstanceIdFutures.get().get("clientId-StreamThread-1-producer");
final Uuid producerUuid = producerFuture.get();
assertThat(producerUuid, equalTo(producerInstanceId));
}
} }
@Test @Test