consumerRecord : records) {
- Header header = consumerRecord.headers().lastHeader("headerKey");
- if (header != null)
- assertEquals("headerValue", new String(header.value()));
- }
+ Header header = records.get(0).headers().lastHeader("headerKey");
+ assertEquals("headerValue", new String(header.value()));
+
+ // Test the order of headers in a record is preserved when producing and consuming
+ Header[] headers = records.get(0).headers().toArray();
+ assertEquals("headerKey", headers[0].key());
+ assertEquals("headerKey2", headers[1].key());
+ assertEquals("headerKey3", headers[2].key());
+
verifyShareGroupStateTopicRecordsProduced();
}
}
diff --git a/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/producer/ProducerIdExpirationTest.java b/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/producer/ProducerIdExpirationTest.java
index f79b3786253..a9489c88327 100644
--- a/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/producer/ProducerIdExpirationTest.java
+++ b/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/producer/ProducerIdExpirationTest.java
@@ -204,10 +204,10 @@ public class ProducerIdExpirationTest {
// Update the producer ID expiration ms to a very high value.
admin.incrementalAlterConfigs(producerIdExpirationConfig("100000"));
- cluster.brokers().values().forEach(broker -> {
+ cluster.brokers().values().forEach(broker ->
TestUtils.waitUntilTrue(() -> broker.logManager().producerStateManagerConfig().producerIdExpirationMs() == 100000,
- () -> "Configuration was not updated.", DEFAULT_MAX_WAIT_MS, 100);
- });
+ () -> "Configuration was not updated.", DEFAULT_MAX_WAIT_MS, 100)
+ );
// Send more records to send producer ID back to brokers.
producer.send(new ProducerRecord<>(topic1, 0, null, "key".getBytes(), "value".getBytes()));
producer.flush();
@@ -226,10 +226,10 @@ public class ProducerIdExpirationTest {
kafkaBroker.awaitShutdown();
kafkaBroker.startup();
cluster.waitForReadyBrokers();
- cluster.brokers().values().forEach(broker -> {
+ cluster.brokers().values().forEach(broker ->
TestUtils.waitUntilTrue(() -> broker.logManager().producerStateManagerConfig().producerIdExpirationMs() == 100,
- () -> "Configuration was not updated.", DEFAULT_MAX_WAIT_MS, 100);
- });
+ () -> "Configuration was not updated.", DEFAULT_MAX_WAIT_MS, 100)
+ );
// Ensure producer ID expires quickly again.
waitProducerIdExpire(admin);
diff --git a/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/producer/ProducerSendWhileDeletionTest.java b/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/producer/ProducerSendWhileDeletionTest.java
index 4301eecbb9c..aa93431cf63 100644
--- a/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/producer/ProducerSendWhileDeletionTest.java
+++ b/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/producer/ProducerSendWhileDeletionTest.java
@@ -184,9 +184,8 @@ public class ProducerSendWhileDeletionTest {
try (var producer = createProducer()) {
for (int i = 1; i <= numRecords; i++) {
producer.send(new ProducerRecord<>(topic, null, ("value" + i).getBytes()),
- (metadata, exception) -> {
- numAcks.incrementAndGet();
- });
+ (metadata, exception) -> numAcks.incrementAndGet()
+ );
}
producer.flush();
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/AddRaftVoterOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/AddRaftVoterOptions.java
index e28a03d541c..81e889db30d 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/AddRaftVoterOptions.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/AddRaftVoterOptions.java
@@ -17,11 +17,20 @@
package org.apache.kafka.clients.admin;
import org.apache.kafka.common.annotation.InterfaceStability;
+import org.apache.kafka.common.protocol.Errors;
import java.util.Optional;
/**
* Options for {@link Admin#addRaftVoter}.
+ *
+ *
+ * The clusterId is optional.
+ *
+ * If provided, the request will only succeed if the cluster id matches the id of the current cluster.
+ * If the cluster id does not match, the request will fail with {@link Errors#INCONSISTENT_CLUSTER_ID}.
+ *
+ * If not provided, the cluster id check is skipped.
*/
@InterfaceStability.Stable
public class AddRaftVoterOptions extends AbstractOptions {
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/Admin.java b/clients/src/main/java/org/apache/kafka/clients/admin/Admin.java
index e596754f62f..06ede9f620d 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/Admin.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/Admin.java
@@ -32,6 +32,7 @@ import org.apache.kafka.common.acl.AclBindingFilter;
import org.apache.kafka.common.annotation.InterfaceStability;
import org.apache.kafka.common.config.ConfigResource;
import org.apache.kafka.common.errors.FeatureUpdateFailedException;
+import org.apache.kafka.common.errors.InconsistentClusterIdException;
import org.apache.kafka.common.errors.InterruptException;
import org.apache.kafka.common.metrics.KafkaMetric;
import org.apache.kafka.common.quota.ClientQuotaAlteration;
@@ -1866,10 +1867,17 @@ public interface Admin extends AutoCloseable {
/**
* Add a new voter node to the KRaft metadata quorum.
*
+ *
+ * The clusterId in {@link AddRaftVoterOptions} is optional.
+ * If provided, the operation will only succeed if the cluster id matches the id
+ * of the current cluster. If the cluster id does not match, the operation
+ * will fail with {@link InconsistentClusterIdException}.
+ * If not provided, the cluster id check is skipped.
+ *
* @param voterId The node ID of the voter.
* @param voterDirectoryId The directory ID of the voter.
* @param endpoints The endpoints that the new voter has.
- * @param options The options to use when adding the new voter node.
+ * @param options Additional options for the operation, including optional cluster ID.
*/
AddRaftVoterResult addRaftVoter(
int voterId,
@@ -1894,9 +1902,16 @@ public interface Admin extends AutoCloseable {
/**
* Remove a voter node from the KRaft metadata quorum.
*
+ *
+ * The clusterId in {@link AddRaftVoterOptions} is optional.
+ * If provided, the operation will only succeed if the cluster id matches the id
+ * of the current cluster. If the cluster id does not match, the operation
+ * will fail with {@link InconsistentClusterIdException}.
+ * If not provided, the cluster id check is skipped.
+ *
* @param voterId The node ID of the voter.
* @param voterDirectoryId The directory ID of the voter.
- * @param options The options to use when removing the voter node.
+ * @param options Additional options for the operation, including optional cluster ID.
*/
RemoveRaftVoterResult removeRaftVoter(
int voterId,
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/DeleteShareGroupsOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/DeleteShareGroupsOptions.java
index a41ec6d00b3..80fd55c7323 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/DeleteShareGroupsOptions.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/DeleteShareGroupsOptions.java
@@ -22,7 +22,7 @@ import org.apache.kafka.common.annotation.InterfaceStability;
import java.util.Collection;
/**
- * Options for the {@link Admin#deleteShareGroups(Collection , DeleteShareGroupsOptions)} call.
+ * Options for the {@link Admin#deleteShareGroups(Collection, DeleteShareGroupsOptions)} call.
*
* The API of this class is evolving, see {@link Admin} for details.
*/
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/DeleteShareGroupsResult.java b/clients/src/main/java/org/apache/kafka/clients/admin/DeleteShareGroupsResult.java
index c2791e681f7..ff53da08df8 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/DeleteShareGroupsResult.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/DeleteShareGroupsResult.java
@@ -25,7 +25,7 @@ import java.util.HashMap;
import java.util.Map;
/**
- * The result of the {@link Admin#deleteShareGroups(Collection , DeleteShareGroupsOptions)} call.
+ * The result of the {@link Admin#deleteShareGroups(Collection, DeleteShareGroupsOptions)} call.
*
* The API of this class is evolving, see {@link Admin} for details.
*/
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/DeleteStreamsGroupsOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/DeleteStreamsGroupsOptions.java
index 6cd14797122..6ca2ec66a27 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/DeleteStreamsGroupsOptions.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/DeleteStreamsGroupsOptions.java
@@ -21,7 +21,7 @@ import org.apache.kafka.common.annotation.InterfaceStability;
import java.util.Collection;
/**
- * Options for the {@link Admin#deleteStreamsGroups(Collection, DeleteStreamsGroupsOptions)} call.
+ * Options for the {@link Admin#deleteStreamsGroups(Collection, DeleteStreamsGroupsOptions)} call.
*
* The API of this class is evolving, see {@link Admin} for details.
*/
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
index 90f83eac935..78a7f905319 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
@@ -2494,8 +2494,7 @@ public class KafkaAdminClient extends AdminClient {
DescribeClusterResponse response = (DescribeClusterResponse) abstractResponse;
Errors error = Errors.forCode(response.data().errorCode());
if (error != Errors.NONE) {
- ApiError apiError = new ApiError(error, response.data().errorMessage());
- handleFailure(apiError.exception());
+ handleFailure(error.exception(response.data().errorMessage()));
return;
}
@@ -2691,10 +2690,9 @@ public class KafkaAdminClient extends AdminClient {
} else {
List filterResults = new ArrayList<>();
for (DeleteAclsMatchingAcl matchingAcl : filterResult.matchingAcls()) {
- ApiError aclError = new ApiError(Errors.forCode(matchingAcl.errorCode()),
- matchingAcl.errorMessage());
+ Errors aclError = Errors.forCode(matchingAcl.errorCode());
AclBinding aclBinding = DeleteAclsResponse.aclBinding(matchingAcl);
- filterResults.add(new FilterResult(aclBinding, aclError.exception()));
+ filterResults.add(new FilterResult(aclBinding, aclError.exception(matchingAcl.errorMessage())));
}
future.complete(new FilterResults(filterResults));
}
@@ -3995,7 +3993,7 @@ public class KafkaAdminClient extends AdminClient {
for (ReassignablePartitionResponse partition : topicResponse.partitions()) {
errors.put(
new TopicPartition(topicName, partition.partitionIndex()),
- new ApiError(topLevelError, response.data().errorMessage()).exception()
+ topLevelError.exception(response.data().errorMessage())
);
receivedResponsesCount += 1;
}
@@ -4035,7 +4033,7 @@ public class KafkaAdminClient extends AdminClient {
if (partitionError == Errors.NONE) {
errors.put(tp, null);
} else {
- errors.put(tp, new ApiError(partitionError, partResponse.errorMessage()).exception());
+ errors.put(tp, partitionError.exception(partResponse.errorMessage()));
}
receivedResponsesCount += 1;
}
@@ -4111,7 +4109,7 @@ public class KafkaAdminClient extends AdminClient {
handleNotControllerError(error);
break;
default:
- partitionReassignmentsFuture.completeExceptionally(new ApiError(error, response.data().errorMessage()).exception());
+ partitionReassignmentsFuture.completeExceptionally(error.exception(response.data().errorMessage()));
break;
}
Map reassignmentMap = new HashMap<>();
@@ -4993,14 +4991,11 @@ public class KafkaAdminClient extends AdminClient {
void handleResponse(AbstractResponse response) {
handleNotControllerError(response);
AddRaftVoterResponse addResponse = (AddRaftVoterResponse) response;
- if (addResponse.data().errorCode() != Errors.NONE.code()) {
- ApiError error = new ApiError(
- addResponse.data().errorCode(),
- addResponse.data().errorMessage());
- future.completeExceptionally(error.exception());
- } else {
+ Errors error = Errors.forCode(addResponse.data().errorCode());
+ if (error != Errors.NONE)
+ future.completeExceptionally(error.exception(addResponse.data().errorMessage()));
+ else
future.complete(null);
- }
}
@Override
@@ -5038,14 +5033,11 @@ public class KafkaAdminClient extends AdminClient {
void handleResponse(AbstractResponse response) {
handleNotControllerError(response);
RemoveRaftVoterResponse addResponse = (RemoveRaftVoterResponse) response;
- if (addResponse.data().errorCode() != Errors.NONE.code()) {
- ApiError error = new ApiError(
- addResponse.data().errorCode(),
- addResponse.data().errorMessage());
- future.completeExceptionally(error.exception());
- } else {
+ Errors error = Errors.forCode(addResponse.data().errorCode());
+ if (error != Errors.NONE)
+ future.completeExceptionally(error.exception(addResponse.data().errorMessage()));
+ else
future.complete(null);
- }
}
@Override
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/ListShareGroupOffsetsResult.java b/clients/src/main/java/org/apache/kafka/clients/admin/ListShareGroupOffsetsResult.java
index 6fecddc69ef..1a2c8869c6c 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/ListShareGroupOffsetsResult.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/ListShareGroupOffsetsResult.java
@@ -46,7 +46,7 @@ public class ListShareGroupOffsetsResult {
/**
* Return the future when the requests for all groups succeed.
*
- * @return Future which yields all {@code Map>} objects, if requests for all the groups succeed.
+ * @return Future which yields all {@code Map>} objects, if requests for all the groups succeed.
*/
public KafkaFuture>> all() {
return KafkaFuture.allOf(futures.values().toArray(new KafkaFuture[0])).thenApply(
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/ListStreamsGroupOffsetsSpec.java b/clients/src/main/java/org/apache/kafka/clients/admin/ListStreamsGroupOffsetsSpec.java
index ce2d1552a53..4f5380f7491 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/ListStreamsGroupOffsetsSpec.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/ListStreamsGroupOffsetsSpec.java
@@ -21,6 +21,7 @@ import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.annotation.InterfaceStability;
import java.util.Collection;
+import java.util.Map;
/**
* Specification of streams group offsets to list using {@link Admin#listStreamsGroupOffsets(Map, ListStreamsGroupOffsetsOptions)}.
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/RecordsToDelete.java b/clients/src/main/java/org/apache/kafka/clients/admin/RecordsToDelete.java
index d3da26b03bb..57421e3568b 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/RecordsToDelete.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/RecordsToDelete.java
@@ -33,14 +33,16 @@ public class RecordsToDelete {
/**
* Delete all the records before the given {@code offset}
*
- * @param offset the offset before which all records will be deleted
+ * @param offset The offset before which all records will be deleted.
+ * Use {@code -1} to truncate to the high watermark.
*/
public static RecordsToDelete beforeOffset(long offset) {
return new RecordsToDelete(offset);
}
/**
- * The offset before which all records will be deleted
+ * The offset before which all records will be deleted.
+ * Use {@code -1} to truncate to the high watermark.
*/
public long beforeOffset() {
return offset;
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/RemoveRaftVoterOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/RemoveRaftVoterOptions.java
index cb5fe563c19..da6e965ebe0 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/RemoveRaftVoterOptions.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/RemoveRaftVoterOptions.java
@@ -17,11 +17,20 @@
package org.apache.kafka.clients.admin;
import org.apache.kafka.common.annotation.InterfaceStability;
+import org.apache.kafka.common.protocol.Errors;
import java.util.Optional;
/**
* Options for {@link Admin#removeRaftVoter}.
+ *
+ *
+ * The clusterId is optional.
+ *
+ * If provided, the request will only succeed if the cluster id matches the id of the current cluster.
+ * If the cluster id does not match, the request will fail with {@link Errors#INCONSISTENT_CLUSTER_ID}.
+ *
+ * If not provided, the cluster id check is skipped.
*/
@InterfaceStability.Stable
public class RemoveRaftVoterOptions extends AbstractOptions {
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/internals/DeleteShareGroupOffsetsHandler.java b/clients/src/main/java/org/apache/kafka/clients/admin/internals/DeleteShareGroupOffsetsHandler.java
index 9f5742937c8..9f0ab59e32c 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/internals/DeleteShareGroupOffsetsHandler.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/internals/DeleteShareGroupOffsetsHandler.java
@@ -122,8 +122,8 @@ public class DeleteShareGroupOffsetsHandler extends AdminApiHandler.Batched handleUnsupportedVersionException(
int brokerId, UnsupportedVersionException exception, Set keys
) {
- log.warn("Broker " + brokerId + " does not support MAX_TIMESTAMP offset specs");
+ log.warn("Broker {} does not support MAX_TIMESTAMP offset specs", brokerId);
Map maxTimestampPartitions = new HashMap<>();
for (TopicPartition topicPartition : keys) {
Long offsetTimestamp = offsetTimestampsByPartition.get(topicPartition);
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerRebalanceListener.java b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerRebalanceListener.java
index c49b2c8045a..23e045b7600 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerRebalanceListener.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerRebalanceListener.java
@@ -50,7 +50,7 @@ import java.util.Collection;
* Under normal conditions, if a partition is reassigned from one consumer to another, then the old consumer will
* always invoke {@link #onPartitionsRevoked(Collection) onPartitionsRevoked} for that partition prior to the new consumer
* invoking {@link #onPartitionsAssigned(Collection) onPartitionsAssigned} for the same partition. So if offsets or other state is saved in the
- * {@link #onPartitionsRevoked(Collection) onPartitionsRevoked} call by one consumer member, it will be always accessible by the time the
+ * {@link #onPartitionsRevoked(Collection) onPartitionsRevoked} call by one consumer member, it will always be accessible by the time the
* other consumer member taking over that partition and triggering its {@link #onPartitionsAssigned(Collection) onPartitionsAssigned} callback to load the state.
*
* You can think of revocation as a graceful way to give up ownership of a partition. In some cases, the consumer may not have an opportunity to do so.
@@ -120,13 +120,31 @@ public interface ConsumerRebalanceListener {
/**
* A callback method the user can implement to provide handling of offset commits to a customized store.
* This method will be called during a rebalance operation when the consumer has to give up some partitions.
- * It can also be called when consumer is being closed ({@link KafkaConsumer#close(CloseOptions option)})
- * or is unsubscribing ({@link KafkaConsumer#unsubscribe()}).
+ * The consumer may need to give up some partitions (thus this callback executed) under the following scenarios:
+ *
+ * - If the consumer assignment changes
+ * - If the consumer is being closed ({@link KafkaConsumer#close(CloseOptions option)})
+ * - If the consumer is unsubscribing ({@link KafkaConsumer#unsubscribe()})
+ *
* It is recommended that offsets should be committed in this callback to either Kafka or a
* custom offset store to prevent duplicate data.
*
- * In eager rebalancing, it will always be called at the start of a rebalance and after the consumer stops fetching data.
- * In cooperative rebalancing, it will be called at the end of a rebalance on the set of partitions being revoked iff the set is non-empty.
+ * This callback is always called before re-assigning the partitions.
+ * If the consumer is using the {@link GroupProtocol#CLASSIC} rebalance protocol:
+ *
+ * -
+ * In eager rebalancing, onPartitionsRevoked will be called with the full set of assigned partitions as a parameter (all partitions are revoked).
+ * It will be called even if there are no partitions to revoke.
+ *
+ * -
+ * In cooperative rebalancing, onPartitionsRevoked will be called with the set of partitions to revoke,
+ * iff the set is non-empty.
+ *
+ *
+ * If the consumer is using the {@link GroupProtocol#CONSUMER} rebalance protocol, this callback will be called
+ * with the set of partitions to revoke iff the set is non-empty
+ * (same behavior as the {@link GroupProtocol#CLASSIC} rebalance protocol with Cooperative mode).
+ *
* For examples on usage of this API, see Usage Examples section of {@link KafkaConsumer KafkaConsumer}.
*
* It is common for the revocation callback to use the consumer instance in order to commit offsets. It is possible
@@ -135,8 +153,9 @@ public interface ConsumerRebalanceListener {
* invocation of {@link KafkaConsumer#poll(java.time.Duration)} in which this callback is being executed. This means it is not
* necessary to catch these exceptions and re-attempt to wakeup or interrupt the consumer thread.
*
- * @param partitions The list of partitions that were assigned to the consumer and now need to be revoked (may not
- * include all currently assigned partitions, i.e. there may still be some partitions left)
+ * @param partitions The list of partitions that were assigned to the consumer and now need to be revoked. This will
+ * include the full assignment under the Classic/Eager protocol, given that it revokes all partitions.
+ * It will only include the subset to revoke under the Classic/Cooperative and Consumer protocols.
* @throws org.apache.kafka.common.errors.WakeupException If raised from a nested call to {@link KafkaConsumer}
* @throws org.apache.kafka.common.errors.InterruptException If raised from a nested call to {@link KafkaConsumer}
*/
@@ -144,12 +163,13 @@ public interface ConsumerRebalanceListener {
/**
* A callback method the user can implement to provide handling of customized offsets on completion of a successful
- * partition re-assignment. This method will be called after the partition re-assignment completes and before the
- * consumer starts fetching data, and only as the result of a {@link Consumer#poll(java.time.Duration) poll(long)} call.
+ * partition re-assignment. This method will be called after the partition re-assignment completes (even if no new
+ * partitions were assigned to the consumer), and before the consumer starts fetching data,
+ * and only as the result of a {@link Consumer#poll(java.time.Duration) poll(long)} call.
*
* It is guaranteed that under normal conditions all the processes in a consumer group will execute their
- * {@link #onPartitionsRevoked(Collection)} callback before any instance executes its
- * {@link #onPartitionsAssigned(Collection)} callback. During exceptional scenarios, partitions may be migrated
+ * {@link #onPartitionsRevoked(Collection)} callback before any instance executes this onPartitionsAssigned callback.
+ * During exceptional scenarios, partitions may be migrated
* without the old owner being notified (i.e. their {@link #onPartitionsRevoked(Collection)} callback not triggered),
* and later when the old owner consumer realized this event, the {@link #onPartitionsLost(Collection)} callback
* will be triggered by the consumer then.
@@ -160,9 +180,11 @@ public interface ConsumerRebalanceListener {
* invocation of {@link KafkaConsumer#poll(java.time.Duration)} in which this callback is being executed. This means it is not
* necessary to catch these exceptions and re-attempt to wakeup or interrupt the consumer thread.
*
- * @param partitions The list of partitions that are now assigned to the consumer (previously owned partitions will
- * NOT be included, i.e. this list will only include newly added partitions)
- * @throws org.apache.kafka.common.errors.WakeupException If raised from a nested call to {@link KafkaConsumer}
+ * @param partitions Partitions that have been added to the assignment as a result of the rebalance.
+ * Note that partitions that were already owned by this consumer and remain assigned are not
+ * included in this list under the Classic/Cooperative or Consumer protocols. THe full assignment
+ * will be received under the Classic/Eager protocol.
+ * @throws org.apache.kafka.common.errors.WakeupException If raised from a nested call to {@link KafkaConsumer}
* @throws org.apache.kafka.common.errors.InterruptException If raised from a nested call to {@link KafkaConsumer}
*/
void onPartitionsAssigned(Collection partitions);
@@ -187,10 +209,9 @@ public interface ConsumerRebalanceListener {
* necessary to catch these exceptions and re-attempt to wakeup or interrupt the consumer thread.
*
* @param partitions The list of partitions that were assigned to the consumer and now have been reassigned
- * to other consumers. With the current protocol this will always include all of the consumer's
- * previously assigned partitions, but this may change in future protocols (ie there would still
- * be some partitions left)
- * @throws org.apache.kafka.common.errors.WakeupException If raised from a nested call to {@link KafkaConsumer}
+ * to other consumers. With both, the Classic and Consumer protocols, this will always include
+ * all partitions that were previously assigned to the consumer.
+ * @throws org.apache.kafka.common.errors.WakeupException If raised from a nested call to {@link KafkaConsumer}
* @throws org.apache.kafka.common.errors.InterruptException If raised from a nested call to {@link KafkaConsumer}
*/
default void onPartitionsLost(Collection partitions) {
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
index e74cf0414a8..9f1992d6568 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
@@ -661,7 +661,7 @@ public class KafkaConsumer implements Consumer {
* If the given list of topics is empty, it is treated the same as {@link #unsubscribe()}.
*
*
- * As part of group management, the consumer will keep track of the list of consumers that belong to a particular
+ * As part of group management, the group coordinator will keep track of the list of consumers that belong to a particular
* group and will trigger a rebalance operation if any one of the following events are triggered:
*
* - Number of partitions change for any of the subscribed topics
@@ -670,8 +670,11 @@ public class KafkaConsumer implements Consumer {
*
- A new member is added to the consumer group
*
*
- * When any of these events are triggered, the provided listener will be invoked first to indicate that
- * the consumer's assignment has been revoked, and then again when the new assignment has been received.
+ * When any of these events are triggered, the provided listener will be invoked in this way:
+ *
+ * - {@link ConsumerRebalanceListener#onPartitionsRevoked(Collection)} will be invoked with the partitions to revoke, before re-assigning those partitions to another consumer.
+ * - {@link ConsumerRebalanceListener#onPartitionsAssigned(Collection)} will be invoked when the rebalance completes (even if no new partitions are assigned to the consumer)
+ *
* Note that rebalances will only occur during an active call to {@link #poll(Duration)}, so callbacks will
* also only be invoked during that time.
*
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java
index 74ecf7f9bb8..ffe01c089e7 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java
@@ -988,7 +988,7 @@ public abstract class AbstractMembershipManager impl
String reason = rejoinedWhileReconciliationInProgress ?
"the member has re-joined the group" :
"the member already transitioned out of the reconciling state into " + state;
- log.info("Interrupting reconciliation that is not relevant anymore because " + reason);
+ log.info("Interrupting reconciliation that is not relevant anymore because {}", reason);
markReconciliationCompleted();
}
return shouldAbort;
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignor.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignor.java
index 4ac1513ede5..c38b5859f5f 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignor.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignor.java
@@ -965,8 +965,8 @@ public abstract class AbstractStickyAssignor extends AbstractPartitionAssignor {
super(partitionsPerTopic, rackInfo, currentAssignment);
this.subscriptions = subscriptions;
- topic2AllPotentialConsumers = new HashMap<>(partitionsPerTopic.keySet().size());
- consumer2AllPotentialTopics = new HashMap<>(subscriptions.keySet().size());
+ topic2AllPotentialConsumers = new HashMap<>(partitionsPerTopic.size());
+ consumer2AllPotentialTopics = new HashMap<>(subscriptions.size());
// initialize topic2AllPotentialConsumers and consumer2AllPotentialTopics
partitionsPerTopic.keySet().forEach(
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java
index 5c72c2babbb..938ae909027 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java
@@ -187,25 +187,6 @@ public class AsyncKafkaConsumer implements ConsumerDelegate {
*/
private class BackgroundEventProcessor implements EventProcessor {
- private Optional streamsRebalanceListener = Optional.empty();
- private final Optional streamsRebalanceData;
-
- public BackgroundEventProcessor() {
- this.streamsRebalanceData = Optional.empty();
- }
-
- public BackgroundEventProcessor(final Optional streamsRebalanceData) {
- this.streamsRebalanceData = streamsRebalanceData;
- }
-
- private void setStreamsRebalanceListener(final StreamsRebalanceListener streamsRebalanceListener) {
- if (streamsRebalanceData.isEmpty()) {
- throw new IllegalStateException("Background event processor was not created to be used with Streams " +
- "rebalance protocol events");
- }
- this.streamsRebalanceListener = Optional.of(streamsRebalanceListener);
- }
-
@Override
public void process(final BackgroundEvent event) {
switch (event.type()) {
@@ -278,44 +259,26 @@ public class AsyncKafkaConsumer implements ConsumerDelegate {
private StreamsOnTasksRevokedCallbackCompletedEvent invokeOnTasksRevokedCallback(final Set activeTasksToRevoke,
final CompletableFuture future) {
- final Optional exceptionFromCallback = streamsRebalanceListener().onTasksRevoked(activeTasksToRevoke);
+ final Optional exceptionFromCallback = Optional.ofNullable(streamsRebalanceListenerInvoker().invokeTasksRevoked(activeTasksToRevoke));
final Optional error = exceptionFromCallback.map(e -> ConsumerUtils.maybeWrapAsKafkaException(e, "Task revocation callback throws an error"));
return new StreamsOnTasksRevokedCallbackCompletedEvent(future, error);
}
private StreamsOnTasksAssignedCallbackCompletedEvent invokeOnTasksAssignedCallback(final StreamsRebalanceData.Assignment assignment,
final CompletableFuture future) {
- final Optional error;
- final Optional exceptionFromCallback = streamsRebalanceListener().onTasksAssigned(assignment);
- if (exceptionFromCallback.isPresent()) {
- error = Optional.of(ConsumerUtils.maybeWrapAsKafkaException(exceptionFromCallback.get(), "Task assignment callback throws an error"));
- } else {
- error = Optional.empty();
- streamsRebalanceData().setReconciledAssignment(assignment);
- }
+ final Optional exceptionFromCallback = Optional.ofNullable(streamsRebalanceListenerInvoker().invokeTasksAssigned(assignment));
+ final Optional error = exceptionFromCallback.map(e -> ConsumerUtils.maybeWrapAsKafkaException(e, "Task assignment callback throws an error"));
return new StreamsOnTasksAssignedCallbackCompletedEvent(future, error);
}
private StreamsOnAllTasksLostCallbackCompletedEvent invokeOnAllTasksLostCallback(final CompletableFuture future) {
- final Optional error;
- final Optional exceptionFromCallback = streamsRebalanceListener().onAllTasksLost();
- if (exceptionFromCallback.isPresent()) {
- error = Optional.of(ConsumerUtils.maybeWrapAsKafkaException(exceptionFromCallback.get(), "All tasks lost callback throws an error"));
- } else {
- error = Optional.empty();
- streamsRebalanceData().setReconciledAssignment(StreamsRebalanceData.Assignment.EMPTY);
- }
+ final Optional exceptionFromCallback = Optional.ofNullable(streamsRebalanceListenerInvoker().invokeAllTasksLost());
+ final Optional error = exceptionFromCallback.map(e -> ConsumerUtils.maybeWrapAsKafkaException(e, "All tasks lost callback throws an error"));
return new StreamsOnAllTasksLostCallbackCompletedEvent(future, error);
}
- private StreamsRebalanceData streamsRebalanceData() {
- return streamsRebalanceData.orElseThrow(
- () -> new IllegalStateException("Background event processor was not created to be used with Streams " +
- "rebalance protocol events"));
- }
-
- private StreamsRebalanceListener streamsRebalanceListener() {
- return streamsRebalanceListener.orElseThrow(
+ private StreamsRebalanceListenerInvoker streamsRebalanceListenerInvoker() {
+ return streamsRebalanceListenerInvoker.orElseThrow(
() -> new IllegalStateException("Background event processor was not created to be used with Streams " +
"rebalance protocol events"));
}
@@ -367,6 +330,7 @@ public class AsyncKafkaConsumer implements ConsumerDelegate {
private final WakeupTrigger wakeupTrigger = new WakeupTrigger();
private final OffsetCommitCallbackInvoker offsetCommitCallbackInvoker;
private final ConsumerRebalanceListenerInvoker rebalanceListenerInvoker;
+ private final Optional streamsRebalanceListenerInvoker;
// Last triggered async commit future. Used to wait until all previous async commits are completed.
// We only need to keep track of the last one, since they are guaranteed to complete in order.
private CompletableFuture> lastPendingAsyncCommit = null;
@@ -517,7 +481,9 @@ public class AsyncKafkaConsumer implements ConsumerDelegate {
time,
new RebalanceCallbackMetricsManager(metrics)
);
- this.backgroundEventProcessor = new BackgroundEventProcessor(streamsRebalanceData);
+ this.streamsRebalanceListenerInvoker = streamsRebalanceData.map(s ->
+ new StreamsRebalanceListenerInvoker(logContext, s));
+ this.backgroundEventProcessor = new BackgroundEventProcessor();
this.backgroundEventReaper = backgroundEventReaperFactory.build(logContext);
// The FetchCollector is only used on the application thread.
@@ -577,6 +543,7 @@ public class AsyncKafkaConsumer implements ConsumerDelegate {
this.time = time;
this.backgroundEventQueue = backgroundEventQueue;
this.rebalanceListenerInvoker = rebalanceListenerInvoker;
+ this.streamsRebalanceListenerInvoker = Optional.empty();
this.backgroundEventProcessor = new BackgroundEventProcessor();
this.backgroundEventReaper = backgroundEventReaper;
this.metrics = metrics;
@@ -699,6 +666,7 @@ public class AsyncKafkaConsumer implements ConsumerDelegate {
networkClientDelegateSupplier,
requestManagersSupplier,
asyncConsumerMetrics);
+ this.streamsRebalanceListenerInvoker = Optional.empty();
this.backgroundEventProcessor = new BackgroundEventProcessor();
this.backgroundEventReaper = new CompletableEventReaper(logContext);
}
@@ -1477,7 +1445,7 @@ public class AsyncKafkaConsumer implements ConsumerDelegate {
() -> autoCommitOnClose(closeTimer), firstException);
swallow(log, Level.ERROR, "Failed to stop finding coordinator",
this::stopFindCoordinatorOnClose, firstException);
- swallow(log, Level.ERROR, "Failed to release group assignment",
+ swallow(log, Level.ERROR, "Failed to run rebalance callbacks",
this::runRebalanceCallbacksOnClose, firstException);
swallow(log, Level.ERROR, "Failed to leave group while closing consumer",
() -> leaveGroupOnClose(closeTimer, membershipOperation), firstException);
@@ -1527,26 +1495,39 @@ public class AsyncKafkaConsumer implements ConsumerDelegate {
}
private void runRebalanceCallbacksOnClose() {
- if (groupMetadata.get().isEmpty() || rebalanceListenerInvoker == null)
+ if (groupMetadata.get().isEmpty())
return;
int memberEpoch = groupMetadata.get().get().generationId();
- Set assignedPartitions = groupAssignmentSnapshot.get();
+ Exception error = null;
- if (assignedPartitions.isEmpty())
- // Nothing to revoke.
- return;
+ if (streamsRebalanceListenerInvoker != null && streamsRebalanceListenerInvoker.isPresent()) {
- SortedSet droppedPartitions = new TreeSet<>(TOPIC_PARTITION_COMPARATOR);
- droppedPartitions.addAll(assignedPartitions);
+ if (memberEpoch > 0) {
+ error = streamsRebalanceListenerInvoker.get().invokeAllTasksRevoked();
+ } else {
+ error = streamsRebalanceListenerInvoker.get().invokeAllTasksLost();
+ }
- final Exception error;
+ } else if (rebalanceListenerInvoker != null) {
- if (memberEpoch > 0)
- error = rebalanceListenerInvoker.invokePartitionsRevoked(droppedPartitions);
- else
- error = rebalanceListenerInvoker.invokePartitionsLost(droppedPartitions);
+ Set assignedPartitions = groupAssignmentSnapshot.get();
+
+ if (assignedPartitions.isEmpty())
+ // Nothing to revoke.
+ return;
+
+ SortedSet droppedPartitions = new TreeSet<>(TOPIC_PARTITION_COMPARATOR);
+ droppedPartitions.addAll(assignedPartitions);
+
+ if (memberEpoch > 0) {
+ error = rebalanceListenerInvoker.invokePartitionsRevoked(droppedPartitions);
+ } else {
+ error = rebalanceListenerInvoker.invokePartitionsLost(droppedPartitions);
+ }
+
+ }
if (error != null)
throw ConsumerUtils.maybeWrapAsKafkaException(error);
@@ -1963,8 +1944,12 @@ public class AsyncKafkaConsumer implements ConsumerDelegate {
}
public void subscribe(Collection topics, StreamsRebalanceListener streamsRebalanceListener) {
+
+ streamsRebalanceListenerInvoker
+ .orElseThrow(() -> new IllegalStateException("Consumer was not created to be used with Streams rebalance protocol events"))
+ .setRebalanceListener(streamsRebalanceListener);
+
subscribeInternal(topics, Optional.empty());
- backgroundEventProcessor.setStreamsRebalanceListener(streamsRebalanceListener);
}
@Override
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java
index b6a6a9f14cf..6aae084fd47 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java
@@ -181,6 +181,14 @@ public class CommitRequestManager implements RequestManager, MemberStateListener
// poll when the coordinator node is known and fatal error is not present
if (coordinatorRequestManager.coordinator().isEmpty()) {
pendingRequests.maybeFailOnCoordinatorFatalError();
+
+ if (closing && pendingRequests.hasUnsentRequests()) {
+ CommitFailedException exception = new CommitFailedException(
+ "Failed to commit offsets: Coordinator unknown and consumer is closing");
+ pendingRequests.drainPendingCommits()
+ .forEach(request -> request.future().completeExceptionally(exception));
+ }
+
return EMPTY;
}
@@ -541,7 +549,7 @@ public class CommitRequestManager implements RequestManager, MemberStateListener
boolean inflightRemoved = pendingRequests.inflightOffsetFetches.remove(fetchRequest);
if (!inflightRemoved) {
log.warn("A duplicated, inflight, request was identified, but unable to find it in the " +
- "outbound buffer:" + fetchRequest);
+ "outbound buffer: {}", fetchRequest);
}
if (error == null) {
maybeUpdateLastSeenEpochIfNewer(res);
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerMembershipManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerMembershipManager.java
index e07424a6393..25e523c3a0d 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerMembershipManager.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerMembershipManager.java
@@ -163,7 +163,7 @@ public class ConsumerMembershipManager extends AbstractMembershipManager partitionsToAssign = topicPartitionsForActiveTasks(activeTasksToAssign);
- final SortedSet partitionsToAssigneNotPreviouslyOwned =
+ final SortedSet partitionsToAssignNotPreviouslyOwned =
partitionsToAssignNotPreviouslyOwned(partitionsToAssign, topicPartitionsForActiveTasks(ownedActiveTasks));
subscriptionState.assignFromSubscribedAwaitingCallback(
partitionsToAssign,
- partitionsToAssigneNotPreviouslyOwned
+ partitionsToAssignNotPreviouslyOwned
);
notifyAssignmentChange(partitionsToAssign);
@@ -1152,10 +1152,10 @@ public class StreamsMembershipManager implements RequestManager {
if (callbackError == null) {
subscriptionState.enablePartitionsAwaitingCallback(partitionsToAssign);
} else {
- if (!partitionsToAssigneNotPreviouslyOwned.isEmpty()) {
+ if (!partitionsToAssignNotPreviouslyOwned.isEmpty()) {
log.warn("Leaving newly assigned partitions {} marked as non-fetchable and not " +
"requiring initializing positions after onTasksAssigned callback failed.",
- partitionsToAssigneNotPreviouslyOwned, callbackError);
+ partitionsToAssignNotPreviouslyOwned, callbackError);
}
}
});
@@ -1205,9 +1205,9 @@ public class StreamsMembershipManager implements RequestManager {
Stream.concat(
streamsRebalanceData.subtopologies().get(task.subtopologyId()).sourceTopics().stream(),
streamsRebalanceData.subtopologies().get(task.subtopologyId()).repartitionSourceTopics().keySet().stream()
- ).forEach(topic -> {
- topicPartitions.add(new TopicPartition(topic, task.partitionId()));
- })
+ ).forEach(topic ->
+ topicPartitions.add(new TopicPartition(topic, task.partitionId()))
+ )
);
return topicPartitions;
}
@@ -1223,7 +1223,7 @@ public class StreamsMembershipManager implements RequestManager {
String reason = rejoinedWhileReconciliationInProgress ?
"the member has re-joined the group" :
"the member already transitioned out of the reconciling state into " + state;
- log.info("Interrupting reconciliation that is not relevant anymore because " + reason);
+ log.info("Interrupting reconciliation that is not relevant anymore because {}", reason);
markReconciliationCompleted();
}
return shouldAbort;
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceListener.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceListener.java
index 55de41d77a5..2c8a7449a72 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceListener.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceListener.java
@@ -16,7 +16,6 @@
*/
package org.apache.kafka.clients.consumer.internals;
-import java.util.Optional;
import java.util.Set;
/**
@@ -28,22 +27,18 @@ public interface StreamsRebalanceListener {
* Called when tasks are revoked from a stream thread.
*
* @param tasks The tasks to be revoked.
- * @return The exception thrown during the callback, if any.
*/
- Optional onTasksRevoked(final Set tasks);
+ void onTasksRevoked(final Set tasks);
/**
* Called when tasks are assigned from a stream thread.
*
* @param assignment The tasks assigned.
- * @return The exception thrown during the callback, if any.
*/
- Optional onTasksAssigned(final StreamsRebalanceData.Assignment assignment);
+ void onTasksAssigned(final StreamsRebalanceData.Assignment assignment);
/**
- * Called when a stream thread loses all assigned tasks.
- *
- * @return The exception thrown during the callback, if any.
+ * Called when a stream thread loses all assigned tasks
*/
- Optional onAllTasksLost();
+ void onAllTasksLost();
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceListenerInvoker.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceListenerInvoker.java
new file mode 100644
index 00000000000..fc8c78c13dc
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceListenerInvoker.java
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer.internals;
+
+import org.apache.kafka.common.errors.InterruptException;
+import org.apache.kafka.common.errors.WakeupException;
+import org.apache.kafka.common.utils.LogContext;
+
+import org.slf4j.Logger;
+
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+
+/**
+ * This class encapsulates the invocation of the callback methods defined in the {@link StreamsRebalanceListener}
+ * interface. When streams group task assignment changes, these methods are invoked. This class wraps those
+ * callback calls with some logging and error handling.
+ */
+public class StreamsRebalanceListenerInvoker {
+
+ private final Logger log;
+
+ private final StreamsRebalanceData streamsRebalanceData;
+ private Optional listener;
+
+ StreamsRebalanceListenerInvoker(LogContext logContext, StreamsRebalanceData streamsRebalanceData) {
+ this.log = logContext.logger(getClass());
+ this.listener = Optional.empty();
+ this.streamsRebalanceData = streamsRebalanceData;
+ }
+
+ public void setRebalanceListener(StreamsRebalanceListener streamsRebalanceListener) {
+ Objects.requireNonNull(streamsRebalanceListener, "StreamsRebalanceListener cannot be null");
+ this.listener = Optional.of(streamsRebalanceListener);
+ }
+
+ public Exception invokeAllTasksRevoked() {
+ if (listener.isEmpty()) {
+ return null;
+ }
+ return invokeTasksRevoked(streamsRebalanceData.reconciledAssignment().activeTasks());
+ }
+
+ public Exception invokeTasksAssigned(final StreamsRebalanceData.Assignment assignment) {
+ if (listener.isEmpty()) {
+ return null;
+ }
+ log.info("Invoking tasks assigned callback for new assignment: {}", assignment);
+ try {
+ listener.get().onTasksAssigned(assignment);
+ } catch (WakeupException | InterruptException e) {
+ throw e;
+ } catch (Exception e) {
+ log.error(
+ "Streams rebalance listener failed on invocation of onTasksAssigned for tasks {}",
+ assignment,
+ e
+ );
+ return e;
+ }
+ return null;
+ }
+
+ public Exception invokeTasksRevoked(final Set tasks) {
+ if (listener.isEmpty()) {
+ return null;
+ }
+ log.info("Invoking task revoked callback for revoked active tasks {}", tasks);
+ try {
+ listener.get().onTasksRevoked(tasks);
+ } catch (WakeupException | InterruptException e) {
+ throw e;
+ } catch (Exception e) {
+ log.error(
+ "Streams rebalance listener failed on invocation of onTasksRevoked for tasks {}",
+ tasks,
+ e
+ );
+ return e;
+ }
+ return null;
+ }
+
+ public Exception invokeAllTasksLost() {
+ if (listener.isEmpty()) {
+ return null;
+ }
+ log.info("Invoking tasks lost callback for all tasks");
+ try {
+ listener.get().onAllTasksLost();
+ } catch (WakeupException | InterruptException e) {
+ throw e;
+ } catch (Exception e) {
+ log.error(
+ "Streams rebalance listener failed on invocation of onTasksLost.",
+ e
+ );
+ return e;
+ }
+ return null;
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java
index 4659f4cf0cd..9d44e98ed39 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java
@@ -477,7 +477,7 @@ public class SubscriptionState {
* Provides the number of assigned partitions in a thread safe manner.
* @return the number of assigned partitions.
*/
- synchronized int numAssignedPartitions() {
+ public synchronized int numAssignedPartitions() {
return this.assignment.size();
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/metrics/ConsumerRebalanceMetricsManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/metrics/ConsumerRebalanceMetricsManager.java
index c312edd54b6..e271dee5261 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/metrics/ConsumerRebalanceMetricsManager.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/metrics/ConsumerRebalanceMetricsManager.java
@@ -16,6 +16,8 @@
*/
package org.apache.kafka.clients.consumer.internals.metrics;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.consumer.internals.SubscriptionState;
import org.apache.kafka.common.MetricName;
import org.apache.kafka.common.metrics.Measurable;
import org.apache.kafka.common.metrics.Metrics;
@@ -27,7 +29,9 @@ import org.apache.kafka.common.metrics.stats.Max;
import org.apache.kafka.common.metrics.stats.Rate;
import org.apache.kafka.common.metrics.stats.WindowedCount;
+import java.util.Collection;
import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
import static org.apache.kafka.clients.consumer.internals.ConsumerUtils.CONSUMER_METRIC_GROUP_PREFIX;
import static org.apache.kafka.clients.consumer.internals.ConsumerUtils.COORDINATOR_METRICS_SUFFIX;
@@ -44,11 +48,14 @@ public final class ConsumerRebalanceMetricsManager extends RebalanceMetricsManag
public final MetricName lastRebalanceSecondsAgo;
public final MetricName failedRebalanceTotal;
public final MetricName failedRebalanceRate;
+ public final MetricName assignedPartitionsCount;
private long lastRebalanceEndMs = -1L;
private long lastRebalanceStartMs = -1L;
+ private final Metrics metrics;
- public ConsumerRebalanceMetricsManager(Metrics metrics) {
+ public ConsumerRebalanceMetricsManager(Metrics metrics, SubscriptionState subscriptions) {
super(CONSUMER_METRIC_GROUP_PREFIX + COORDINATOR_METRICS_SUFFIX);
+ this.metrics = metrics;
rebalanceLatencyAvg = createMetric(metrics, "rebalance-latency-avg",
"The average time in ms taken for a group to complete a rebalance");
@@ -64,6 +71,9 @@ public final class ConsumerRebalanceMetricsManager extends RebalanceMetricsManag
"The total number of failed rebalance events");
failedRebalanceRate = createMetric(metrics, "failed-rebalance-rate-per-hour",
"The number of failed rebalance events per hour");
+ assignedPartitionsCount = createMetric(metrics, "assigned-partitions",
+ "The number of partitions currently assigned to this consumer");
+ registerAssignedPartitionCount(subscriptions);
successfulRebalanceSensor = metrics.sensor("rebalance-latency");
successfulRebalanceSensor.add(rebalanceLatencyAvg, new Avg());
@@ -106,4 +116,15 @@ public final class ConsumerRebalanceMetricsManager extends RebalanceMetricsManag
public boolean rebalanceStarted() {
return lastRebalanceStartMs > lastRebalanceEndMs;
}
+
+ /**
+ * Register metric to track the number of assigned partitions.
+ * It will consider partitions assigned to the consumer
+ * regardless of whether they were assigned via {@link KafkaConsumer#subscribe(Pattern)} or
+ * {@link KafkaConsumer#assign(Collection)}
+ */
+ private void registerAssignedPartitionCount(SubscriptionState subscriptions) {
+ Measurable numParts = (config, now) -> subscriptions.numAssignedPartitions();
+ metrics.addMetric(assignedPartitionsCount, numParts);
+ }
}
\ No newline at end of file
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
index 48be57f262b..6e656f590e4 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
@@ -1027,15 +1027,15 @@ public class KafkaProducer implements Producer {
* expensive callbacks it is recommended to use your own {@link java.util.concurrent.Executor} in the callback body
* to parallelize processing.
*
- * @param record The record to send
+ * @param record The record to send
* @param callback A user-supplied callback to execute when the record has been acknowledged by the server (null
- * indicates no callback)
- *
- * @throws IllegalStateException if a transactional.id has been configured and no transaction has been started, or
- * when send is invoked after producer has been closed.
- * @throws InterruptException If the thread is interrupted while blocked
+ * indicates no callback)
+ * @throws IllegalStateException if a transactional.id has been configured and no transaction has been started, or
+ * when send is invoked after producer has been closed.
+ * @throws TimeoutException if the topic or the partition specified in the record cannot be found in metadata within {@code max.block.ms}
+ * @throws InterruptException If the thread is interrupted while blocked
* @throws SerializationException If the key or value are not valid objects given the configured serializers
- * @throws KafkaException If a Kafka related error occurs that does not belong to the public API exceptions.
+ * @throws KafkaException If a Kafka related error occurs that does not belong to the public API exceptions.
*/
@Override
public Future send(ProducerRecord record, Callback callback) {
@@ -1335,11 +1335,14 @@ public class KafkaProducer implements Producer {
/**
* Get the partition metadata for the given topic. This can be used for custom partitioning.
+ *
+ * This will attempt to refresh metadata until it finds the topic in it, or the configured {@link ProducerConfig#MAX_BLOCK_MS_CONFIG} expires.
+ *
* @throws AuthenticationException if authentication fails. See the exception for more details
- * @throws AuthorizationException if not authorized to the specified topic. See the exception for more details
- * @throws InterruptException if the thread is interrupted while blocked
- * @throws TimeoutException if metadata could not be refreshed within {@code max.block.ms}
- * @throws KafkaException for all Kafka-related exceptions, including the case where this method is called after producer close
+ * @throws AuthorizationException if not authorized to the specified topic. See the exception for more details
+ * @throws InterruptException if the thread is interrupted while blocked
+ * @throws TimeoutException if the topic cannot be found in metadata within {@code max.block.ms}
+ * @throws KafkaException for all Kafka-related exceptions, including the case where this method is called after producer close
*/
@Override
public List partitionsFor(String topic) {
diff --git a/clients/src/main/java/org/apache/kafka/common/header/Header.java b/clients/src/main/java/org/apache/kafka/common/header/Header.java
index 58869b41fb7..e1d0aa00a44 100644
--- a/clients/src/main/java/org/apache/kafka/common/header/Header.java
+++ b/clients/src/main/java/org/apache/kafka/common/header/Header.java
@@ -16,10 +16,23 @@
*/
package org.apache.kafka.common.header;
+/**
+ * A header is a key-value pair.
+ */
public interface Header {
-
+
+ /**
+ * Returns the key of the header.
+ *
+ * @return the header's key; must not be null.
+ */
String key();
+ /**
+ * Returns the value of the header.
+ *
+ * @return the header's value; may be null.
+ */
byte[] value();
-
+
}
diff --git a/clients/src/main/java/org/apache/kafka/common/header/Headers.java b/clients/src/main/java/org/apache/kafka/common/header/Headers.java
index b736cbcabcc..9cce54a5c5c 100644
--- a/clients/src/main/java/org/apache/kafka/common/header/Headers.java
+++ b/clients/src/main/java/org/apache/kafka/common/header/Headers.java
@@ -16,12 +16,18 @@
*/
package org.apache.kafka.common.header;
+
+/**
+ * A mutable ordered collection of {@link Header} objects. Note that multiple headers may have the same {@link Header#key() key}.
+ *
+ * The order of headers is preserved in the order they were added.
+ */
public interface Headers extends Iterable {
/**
* Adds a header (key inside), to the end, returning if the operation succeeded.
*
- * @param header the Header to be added
+ * @param header the Header to be added.
* @return this instance of the Headers, once the header is added.
* @throws IllegalStateException is thrown if headers are in a read-only state.
*/
@@ -30,17 +36,18 @@ public interface Headers extends Iterable {
/**
* Creates and adds a header, to the end, returning if the operation succeeded.
*
- * @param key of the header to be added.
- * @param value of the header to be added.
+ * @param key of the header to be added; must not be null.
+ * @param value of the header to be added; may be null.
* @return this instance of the Headers, once the header is added.
* @throws IllegalStateException is thrown if headers are in a read-only state.
*/
Headers add(String key, byte[] value) throws IllegalStateException;
/**
- * Removes all headers for the given key returning if the operation succeeded.
+ * Removes all headers for the given key returning if the operation succeeded,
+ * while preserving the insertion order of the remaining headers.
*
- * @param key to remove all headers for.
+ * @param key to remove all headers for; must not be null.
* @return this instance of the Headers, once the header is removed.
* @throws IllegalStateException is thrown if headers are in a read-only state.
*/
@@ -49,16 +56,17 @@ public interface Headers extends Iterable {
/**
* Returns just one (the very last) header for the given key, if present.
*
- * @param key to get the last header for.
+ * @param key to get the last header for; must not be null.
* @return this last header matching the given key, returns null if not present.
*/
Header lastHeader(String key);
/**
* Returns all headers for the given key, in the order they were added in, if present.
+ * The iterator does not support {@link java.util.Iterator#remove()}.
*
- * @param key to return the headers for.
- * @return all headers for the given key, in the order they were added in, if NO headers are present an empty iterable is returned.
+ * @param key to return the headers for; must not be null.
+ * @return all headers for the given key, in the order they were added in, if NO headers are present an empty iterable is returned.
*/
Iterable