@FunctionalInterface
public interface Consumer<T> {
    void accept(T t);
}

Consumer<String> printUpperCase = str -> System.out.println(str.toUpperCase());
printUpperCase.accept("hello"); // HELLO
Здесь мы создали Consumer, который принимает строку и выводит ее в верхнем регистре. Метод accept применяется к строке "hello", и результат выводится на экран.

import java.util.Arrays;
import java.util.List;
import java.util.function.Consumer;

public class ConsumerExample {
    public static void main(String[] args) {
        List<String> names = Arrays.asList("Alice", "Bob", "Charlie", "David", "Eve");

        // Создаем Consumer для вывода имен в верхнем регистре
        Consumer<String> printUpperCase = name -> System.out.println(name.toUpperCase());

        // Применяем Consumer к каждому элементу списка
        names.forEach(printUpperCase);
    }
}
В этом примере мы используем Consumer для вывода каждого имени из списка в верхнем регистре. Метод forEach принимает Consumer и применяет его к каждому элементу списка.

Consumer<String> printUpperCase = str -> System.out.println(str.toUpperCase());
Consumer<String> printLowerCase = str -> System.out.println(str.toLowerCase());

// Комбинируем два Consumer
Consumer<String> printBoth = printUpperCase.andThen(printLowerCase);

printBoth.accept("Hello"); // HELLO hello
В этом примере мы создали два Consumer: один выводит строку в верхнем регистре, а другой — в нижнем. Метод andThen объединяет их, и результат выполнения printBoth — это сначала вывод строки в верхнем регистре, а затем в нижнем.

import java.util.Arrays;
import java.util.List;
import java.util.function.Consumer;

public class StreamConsumerExample {
    public static void main(String[] args) {
        List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5);

        // Создаем Consumer для вывода квадратов чисел
        Consumer<Integer> printSquare = number -> System.out.println(number * number);

        // Применяем Consumer к каждому элементу потока
        numbers.stream().forEach(printSquare);
    }
}
В этом примере мы используем Consumer для вывода квадратов чисел из списка. Метод forEach в Stream API принимает Consumer и применяет его к каждому элементу потока.

import java.util.Arrays;
import java.util.List;
import java.util.function.Consumer;
import java.util.function.Predicate;

public class CombinedExample {
    public static void main(String[] args) {
        List<String> names = Arrays.asList("Alice", "Bob", "Charlie", "David", "Eve");

        // Создаем Predicate для фильтрации имен, начинающихся на "A"
        Predicate<String> startsWithA = name -> name.startsWith("A");

        // Создаем Consumer для вывода имен
        Consumer<String> printName = name -> System.out.println(name);

        // Фильтруем и выводим имена, начинающиеся на "A"
        names.stream()
             .filter(startsWithA)
             .forEach(printName); // Alice
    }
}
В этом примере мы используем Predicate для фильтрации имен, начинающихся на "A", и Consumer для вывода отфильтрованных имен.

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import java.util.*;

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("enable.auto.commit", "false"); // Manual commit

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("topic"), new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        // Commit offsets before losing partitions
        commitOffsets(); // Custom method to commit processed offsets
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        // Restore state or seek to offsets
        /* warm-up: e.g., load local state for partitions */
    }
});

try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        // Process records
        consumer.commitAsync(); // Or commitSync for sync
    }
} finally {
    consumer.close();
}