AppConfig has 24 methods (exceeds 20 allowed). Consider refactoring.

Open

public interface AppConfig extends Config {
    @Key("CONSUMER_POLL_TIMEOUT_MS")
    @DefaultValue("9223372036854775807")
    Long getConsumerPollTimeoutMs();

Method job has a Cognitive Complexity of 17 (exceeds 5 allowed). Consider refactoring.

Open

    @Override
    public Status job() {
        offsetState.startTimer();
        try {
            Instant startTime = Instant.now();

• Read up
  Read up

Method job has 49 lines of code (exceeds 25 allowed). Consider refactoring.

Open

    @Override
    public Status job() {
        offsetState.startTimer();
        try {
            Instant startTime = Instant.now();

Method getMappings has 43 lines of code (exceeds 25 allowed). Consider refactoring.

Open

    private Map<String, Object> getMappings(DynamicMessage message, ColumnMapping columnMapping) {
        if (message == null || columnMapping == null || columnMapping.isEmpty()) {
            return new HashMap<>();
        }
        if (message.getUnknownFields().asMap().size() > 0) {

Method push has a Cognitive Complexity of 13 (exceeds 5 allowed). Consider refactoring.

Open

    @Override
    public Status push(Records records) {
        InsertAllResponse response = insertIntoBQ(records.getRecords());
        if (response.hasErrors()) {
            //parse the error records

• Read up
  Read up

Method main has 39 lines of code (exceeds 25 allowed). Consider refactoring.

Open

    public static void main(String[] args) throws IOException, InterruptedException {
        AppConfig appConfig = ConfigFactory.create(AppConfig.class, System.getenv());
        ProtoMappingConfig protoMappingConfig = ConfigFactory.create(ProtoMappingConfig.class, System.getenv());
        BackOffConfig backOffConfig = ConfigFactory.create(BackOffConfig.class, System.getenv());
        StencilConfig stencilConfig = ConfigFactory.create(StencilConfig.class, System.getenv());

Method parseFields has a Cognitive Complexity of 12 (exceeds 5 allowed). Consider refactoring.

Open

    private ProtoField parseFields(ProtoField protoField, String protoSchema, Map<String, Descriptors.Descriptor> allDescriptors,
                                   Map<String, String> typeNameToPackageNameMap, int level) {

        Descriptors.Descriptor currentProto = descriptorCache.fetch(allDescriptors, typeNameToPackageNameMap, protoSchema);
        if (currentProto == null) {

• Read up
  Read up

Method upsertDatasetAndTable has 34 lines of code (exceeds 25 allowed). Consider refactoring.

Open

    private void upsertDatasetAndTable(TableInfo tableInfo) {
        Dataset dataSet = bigquery.getDataset(tableID.getDataset());
        if (dataSet == null || !bigquery.getDataset(tableID.getDataset()).exists()) {
            bigquery.create(
                    Dataset.newBuilder(tableID.getDataset())

Method push has 29 lines of code (exceeds 25 allowed). Consider refactoring.

Open

    @Override
    public Status push(Records records) {
        InsertAllResponse response = insertIntoBQ(records.getRecords());
        if (response.hasErrors()) {
            //parse the error records

Method pushToSink has 26 lines of code (exceeds 25 allowed). Consider refactoring.

Open

    private Status pushToSink(Records poll) {
        Status status;
        try {
            status = sink.push(poll);
            statsClient.count("kafka.batch.records.size," + statsClient.getBqTags(), poll.getSize());

Method upsertDatasetAndTable has a Cognitive Complexity of 9 (exceeds 5 allowed). Consider refactoring.

Open

    private void upsertDatasetAndTable(TableInfo tableInfo) {
        Dataset dataSet = bigquery.getDataset(tableID.getDataset());
        if (dataSet == null || !bigquery.getDataset(tableID.getDataset()).exists()) {
            bigquery.create(
                    Dataset.newBuilder(tableID.getDataset())

• Read up
  Read up

Method parseResponse has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    public BQFilteredResponse parseResponse(final List<Record> records, final InsertAllResponse bqResponse) {
        Map<Record, Set<BQRecordsErrorType>> filtered = parseErrors(records, bqResponse);
        final List<Record> retryableRecords = new ArrayList<>();
        final List<Record> unhandledRecords = new ArrayList<>();
        final List<Record> oobRecords = new ArrayList<>();

• Read up
  Read up

Method convert has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    public List<Record> convert(final Iterable<ConsumerRecord<byte[], byte[]>> messages) throws InvalidProtocolBufferException {
        ArrayList<Record> validRecords = new ArrayList<>();
        ArrayList<Record> invalidRecords = new ArrayList<>();
        for (ConsumerRecord<byte[], byte[]> message : messages) {
            if (message.value() == null) {

• Read up
  Read up

Method parseFields has 5 arguments (exceeds 4 allowed). Consider refactoring.

Open

    private ProtoField parseFields(ProtoField protoField, String protoSchema, Map<String, Descriptors.Descriptor> allDescriptors,
                                   Map<String, String> typeNameToPackageNameMap, int level) {

Method toString has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    @Override
    public String toString() {
        String errString = "InsertStatus:";
        if (cause.getErrors().isEmpty()) {
            return  errString + " NO_ERRORS";

• Read up
  Read up

Method convert has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    public Map<String, String> convert(Method method, String input) {
        Map<String, String> result = new LinkedHashMap<>();
        String[] chunks = input.split(ELEMENT_SEPARATOR, -1);
        for (String chunk : chunks) {
            String[] entry = chunk.split(VALUE_SEPARATOR, -1);

• Read up
  Read up

Method pushToSink has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

    private Status pushToSink(Records poll) {
        Status status;
        try {
            status = sink.push(poll);
            statsClient.count("kafka.batch.records.size," + statsClient.getBqTags(), poll.getSize());

• Read up
  Read up

Method run has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

    @Override
    public void run() {
        log.info("Started worker {}", getClass().getSimpleName());
        Status status;
        do {

• Read up
  Read up

Method generateColumnMappingsJson has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

    private ObjectNode generateColumnMappingsJson(List<ProtoField> fields) {
        if (fields.size() == 0) {
            return JsonNodeFactory.instance.objectNode();
        }

• Read up
  Read up