Repository: kestra-io/kestra
File: jdbc/src/main/java/io/kestra/jdbc/runner/JdbcQueue.java (Java)

Code before:
import org.jooq.*;
import org.jooq.Record;
import org.jooq.exception.DataException;
import org.jooq.impl.DSL;

import java.io.IOException;
import java.time.Duration;
import java.time.ZonedDateTime;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Supplier;

import static io.kestra.core.utils.Rethrow.throwConsumer;
import static io.kestra.core.utils.Rethrow.throwRunnable;

@Slf4j
public abstract class JdbcQueue<T> implements QueueInterface<T> {
    protected static final ObjectMapper MAPPER = JdbcMapper.of();

    private static final int MAX_ASYNC_THREADS = Runtime.getRuntime().availableProcessors();
    private static final Field<Object> KEY_FIELD = AbstractJdbcRepository.field("key");
    private static final Field<Object> VALUE_FIELD = AbstractJdbcRepository.field("value");
    private static final Field<Object> OFFSET_FIELD = AbstractJdbcRepository.field("offset");
    private static final Field<Object> CONSUMER_GROUP_FIELD = AbstractJdbcRepository.field("consumer_group");
    private static final Field<Object> TYPE_FIELD = AbstractJdbcRepository.field("type");

    private final ExecutorService poolExecutor;
    private final ExecutorService asyncPoolExecutor;

    protected final QueueService queueService;

    protected final Class<T> cls;

    protected final JooqDSLContextWrapper dslContextWrapper;

    protected final Configuration configuration;

    protected final MessageProtectionConfiguration messageProtectionConfiguration;

    private final MetricRegistry metricRegistry;

    protected final Table<Record> table;

    protected final JdbcQueueIndexer jdbcQueueIndexer;

    private final boolean immediateRepoll;


Review comment (refactor):
`KEY_FIELD`, and `VALUE_FIELD` could be in `AbstractJdbcRepository` because they are common to all tables. Also, I'm sure some repo re-define those static fields.

Code after:
import org.jooq.*;
import org.jooq.Record;
import org.jooq.exception.DataException;
import org.jooq.impl.DSL;

import java.io.IOException;
import java.time.Duration;
import java.time.ZonedDateTime;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Supplier;

import static io.kestra.core.utils.Rethrow.throwConsumer;
import static io.kestra.core.utils.Rethrow.throwRunnable;
import static io.kestra.jdbc.repository.AbstractJdbcRepository.VALUE_FIELD;

@Slf4j
public abstract class JdbcQueue<T> implements QueueInterface<T> {
    protected static final ObjectMapper MAPPER = JdbcMapper.of();

    private static final int MAX_ASYNC_THREADS = Runtime.getRuntime().availableProcessors();
    private static final Field<Object> KEY_FIELD = AbstractJdbcRepository.field("key");
    private static final Field<Object> OFFSET_FIELD = AbstractJdbcRepository.field("offset");
    private static final Field<Object> CONSUMER_GROUP_FIELD = AbstractJdbcRepository.field("consumer_group");
    private static final Field<Object> TYPE_FIELD = AbstractJdbcRepository.field("type");

    private final ExecutorService poolExecutor;
    private final ExecutorService asyncPoolExecutor;

    protected final QueueService queueService;

    protected final Class<T> cls;

    protected final JooqDSLContextWrapper dslContextWrapper;

    protected final Configuration configuration;

    protected final MessageProtectionConfiguration messageProtectionConfiguration;

    private final MetricRegistry metricRegistry;

    protected final Table<Record> table;

    protected final JdbcQueueIndexer jdbcQueueIndexer;

    private final boolean immediateRepoll;