Move arg validation in KafkaConsumer ahead of `acquireAndEnsureOpen`
@@ -966,13 +966,12 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> {
*/
@Override
public void subscribe(Pattern pattern, ConsumerRebalanceListener listener) {
- acquireAndEnsureOpen();
- try {
if (pattern == null)
throw new IllegalArgumentException("Topic pattern to subscribe to cannot be null");
+ acquireAndEnsureOpen();
+ try {
throwIfNoAssignorsConfigured();
-
log.debug("Subscribed to pattern: {}", pattern);
this.subscriptions.subscribe(pattern, listener);
this.metadata.needMetadataForAllTopics(true);
@@ -1337,11 +1336,11 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> {
*/
@Override
public void seek(TopicPartition partition, long offset) {
- acquireAndEnsureOpen();
- try {
if (offset < 0)
throw new IllegalArgumentException("seek offset must not be a negative number");
+ acquireAndEnsureOpen();
+ try {
log.debug("Seeking to offset {} for partition {}", offset, partition);
this.subscriptions.seek(partition, offset);
} finally {
@@ -1357,11 +1356,11 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> {
* @throws IllegalArgumentException if {@code partitions} is {@code null} or the provided TopicPartition is not assigned to this consumer
*/
public void seekToBeginning(Collection<TopicPartition> partitions) {
+ if (partitions == null)
+ throw new IllegalArgumentException("Partitions collection cannot be null");
+
acquireAndEnsureOpen();
try {
- if (partitions == null) {
- throw new IllegalArgumentException("Partitions collection cannot be null");
- }
Collection<TopicPartition> parts = partitions.size() == 0 ? this.subscriptions.assignedPartitions() : partitions;
for (TopicPartition tp : parts) {
log.debug("Seeking to beginning of partition {}", tp);
@@ -1383,11 +1382,11 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> {
* @throws IllegalArgumentException if {@code partitions} is {@code null} or the provided TopicPartition is not assigned to this consumer
*/
public void seekToEnd(Collection<TopicPartition> partitions) {
+ if (partitions == null)
+ throw new IllegalArgumentException("Partitions collection cannot be null");
+
acquireAndEnsureOpen();
try {
- if (partitions == null) {
- throw new IllegalArgumentException("Partitions collection cannot be null");
- }
Collection<TopicPartition> parts = partitions.size() == 0 ? this.subscriptions.assignedPartitions() : partitions;
for (TopicPartition tp : parts) {
log.debug("Seeking to end of partition {}", tp);
