@SuppressWarnings("unchecked")
    public boolean process(final long wallClockTime) {
        if (record == null) {
            if (!isProcessable(wallClockTime)) {
                return false;
            }

            // get the next record to process
            record = partitionGroup.nextRecord(recordInfo, wallClockTime);

            // if there is no record to process, return immediately
            if (record == null) {
                return false;
            }
        }

        try {
            final TopicPartition partition = recordInfo.partition();

            if (!(record instanceof CorruptedRecord)) {
                doProcess(wallClockTime);
            }

            // update the consumed offset map after processing is done
            consumedOffsets.put(partition, record.offset());
            commitNeeded = true;

            // after processing this record, if its partition queue's buffered size has been
            // decreased to the threshold, we can then resume the consumption on this partition
            if (recordInfo.queue().size() == maxBufferedSize) {
                partitionsToResume.add(partition);
            }

            record = null;
        } catch (final TimeoutException timeoutException) {
            if (!eosEnabled) {
                throw timeoutException;
            } else {
                record = null;
                throw new TaskCorruptedException(Collections.singleton(id));
            }
        } catch (final FailedProcessingException failedProcessingException) {
            // Do not keep the failed processing exception in the stack trace
            handleException(failedProcessingException.getCause());
        } catch (final StreamsException exception) {
            record = null;
            throw exception;
        } catch (final Exception e) {
            handleException(e);
        } finally {
            processorContext.setCurrentNode(null);
        }

        return true;
    }
Process one record.

@return true if this method processes a record, false if it does not process a record.
@throws TaskMigratedException if the task producer got fenced (EOS only)
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java