diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingAggregationsTranslators.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingAggregationsTranslators.java
index 1579a3d4affa..1683ced890c7 100644
--- a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingAggregationsTranslators.java
+++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingAggregationsTranslators.java
@@ -17,6 +17,11 @@
  */
 package org.apache.beam.runners.flink;
 
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
 import org.apache.beam.runners.core.KeyedWorkItem;
 import org.apache.beam.runners.core.SystemReduceFn;
 import org.apache.beam.runners.core.construction.SerializablePipelineOptions;
@@ -58,14 +63,9 @@
 import org.apache.flink.streaming.api.transformations.TwoInputTransformation;
 import org.apache.flink.util.Collector;
 
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
 public class FlinkStreamingAggregationsTranslators {
-  public static class ConcatenateAsIterable<T> extends Combine.CombineFn<T, Iterable<T>, Iterable<T>> {
+  public static class ConcatenateAsIterable<T>
+      extends Combine.CombineFn<T, Iterable<T>, Iterable<T>> {
     @Override
     public Iterable<T> createAccumulator() {
       return new ArrayList<>();
@@ -214,8 +214,7 @@ WindowDoFnOperator<K, InputAccumT, OutputAccumT> getWindowedAggregateDoFnOperato
         WindowedValue.getFullCoder(workItemCoder, windowingStrategy.getWindowFn().windowCoder());
 
     // Key selector
-    WorkItemKeySelector<K, InputAccumT> workItemKeySelector =
-        new WorkItemKeySelector<>(keyCoder);
+    WorkItemKeySelector<K, InputAccumT> workItemKeySelector = new WorkItemKeySelector<>(keyCoder);
 
     return new WindowDoFnOperator<>(
         reduceFn,
@@ -257,29 +256,30 @@ WindowDoFnOperator<K, InputAccumT, OutputAccumT> getWindowedAggregateDoFnOperato
   }
 
   private static class FlattenIterable<K, InputT>
-      implements FlatMapFunction<WindowedValue<KV<K, Iterable<Iterable<InputT>>>>, WindowedValue<KV<K, Iterable<InputT>>>> {
+      implements FlatMapFunction<
+          WindowedValue<KV<K, Iterable<Iterable<InputT>>>>,
+          WindowedValue<KV<K, Iterable<InputT>>>> {
     @Override
     public void flatMap(
         WindowedValue<KV<K, Iterable<Iterable<InputT>>>> w,
-        Collector<WindowedValue<KV<K, Iterable<InputT>>>> collector) throws Exception {
-      WindowedValue<KV<K, Iterable<InputT>>> flattened = w.withValue(
-          KV.of(
-              w.getValue().getKey(),
-              Iterables.concat(w.getValue().getValue())));
+        Collector<WindowedValue<KV<K, Iterable<InputT>>>> collector)
+        throws Exception {
+      WindowedValue<KV<K, Iterable<InputT>>> flattened =
+          w.withValue(KV.of(w.getValue().getKey(), Iterables.concat(w.getValue().getValue())));
       collector.collect(flattened);
     }
   }
 
   public static <K, InputT, AccumT, OutputT>
-    SingleOutputStreamOperator<WindowedValue<KV<K, OutputT>>> getBatchCombinePerKeyOperator(
-        FlinkStreamingTranslationContext context,
-        PCollection<KV<K, InputT>> input,
-        Map<Integer, PCollectionView<?>> sideInputTagMapping,
-        List<PCollectionView<?>> sideInputs,
-        Coder<WindowedValue<KV<K, AccumT>>> windowedAccumCoder,
-        CombineFnBase.GlobalCombineFn<InputT, AccumT, ?> combineFn,
-        WindowDoFnOperator<K, AccumT, OutputT> finalDoFnOperator,
-        TypeInformation<WindowedValue<KV<K, OutputT>>> outputTypeInfo){
+      SingleOutputStreamOperator<WindowedValue<KV<K, OutputT>>> getBatchCombinePerKeyOperator(
+          FlinkStreamingTranslationContext context,
+          PCollection<KV<K, InputT>> input,
+          Map<Integer, PCollectionView<?>> sideInputTagMapping,
+          List<PCollectionView<?>> sideInputs,
+          Coder<WindowedValue<KV<K, AccumT>>> windowedAccumCoder,
+          CombineFnBase.GlobalCombineFn<InputT, AccumT, ?> combineFn,
+          WindowDoFnOperator<K, AccumT, OutputT> finalDoFnOperator,
+          TypeInformation<WindowedValue<KV<K, OutputT>>> outputTypeInfo) {
 
     String fullName = FlinkStreamingTransformTranslators.getCurrentTransformName(context);
     DataStream<WindowedValue<KV<K, InputT>>> inputDataStream = context.getInputDataStream(input);
@@ -314,50 +314,55 @@ SingleOutputStreamOperator<WindowedValue<KV<K, OutputT>>> getBatchCombinePerKeyO
     if (sideInputs.isEmpty()) {
       return inputDataStream
           .transform(partialName, partialTypeInfo, partialDoFnOperator)
-          .uid(partialName).name(partialName)
+          .uid(partialName)
+          .name(partialName)
           .keyBy(accumKeySelector)
           .transform(fullName, outputTypeInfo, finalDoFnOperator)
-          .uid(fullName).name(fullName);
+          .uid(fullName)
+          .name(fullName);
     } else {
 
       Tuple2<Map<Integer, PCollectionView<?>>, DataStream<RawUnionValue>> transformSideInputs =
           FlinkStreamingTransformTranslators.transformSideInputs(sideInputs, context);
 
       TwoInputTransformation<
-          WindowedValue<KV<K, InputT>>, RawUnionValue, WindowedValue<KV<K, AccumT>>> rawPartialFlinkTransform =
-          new TwoInputTransformation<>(
-              inputDataStream.getTransformation(),
-              transformSideInputs.f1.broadcast().getTransformation(),
-              partialName,
-              partialDoFnOperator,
-              partialTypeInfo,
-              inputDataStream.getParallelism());
+              WindowedValue<KV<K, InputT>>, RawUnionValue, WindowedValue<KV<K, AccumT>>>
+          rawPartialFlinkTransform =
+              new TwoInputTransformation<>(
+                  inputDataStream.getTransformation(),
+                  transformSideInputs.f1.broadcast().getTransformation(),
+                  partialName,
+                  partialDoFnOperator,
+                  partialTypeInfo,
+                  inputDataStream.getParallelism());
 
       SingleOutputStreamOperator<WindowedValue<KV<K, AccumT>>> partialyCombinedStream =
           new SingleOutputStreamOperator<WindowedValue<KV<K, AccumT>>>(
               inputDataStream.getExecutionEnvironment(),
               rawPartialFlinkTransform) {}; // we have to cheat around the ctor being protected
 
-        inputDataStream.getExecutionEnvironment().addOperator(rawPartialFlinkTransform);
+      inputDataStream.getExecutionEnvironment().addOperator(rawPartialFlinkTransform);
 
-        return buildTwoInputStream(
-            partialyCombinedStream.keyBy(accumKeySelector),
-            transformSideInputs.f1,
-            fullName,
-            finalDoFnOperator,
-            outputTypeInfo);
+      return buildTwoInputStream(
+          partialyCombinedStream.keyBy(accumKeySelector),
+          transformSideInputs.f1,
+          fullName,
+          finalDoFnOperator,
+          outputTypeInfo);
     }
   }
 
   /**
-   * Creates a two-steps GBK operation. Elements are first aggregated locally to save on serialized size since in batch
-   * it's very likely that all the elements will be within the same window and pane.
-   * The only difference with batchCombinePerKey is the nature of the SystemReduceFn used. It uses SystemReduceFn.buffering()
-   * instead of SystemReduceFn.combining() so that new element can simply be appended without accessing the existing state.
+   * Creates a two-steps GBK operation. Elements are first aggregated locally to save on serialized
+   * size since in batch it's very likely that all the elements will be within the same window and
+   * pane. The only difference with batchCombinePerKey is the nature of the SystemReduceFn used. It
+   * uses SystemReduceFn.buffering() instead of SystemReduceFn.combining() so that new element can
+   * simply be appended without accessing the existing state.
    */
-  public static <K, InputT> SingleOutputStreamOperator<WindowedValue<KV<K, Iterable<InputT>>>> batchGroupByKey(
-      FlinkStreamingTranslationContext context,
-      PTransform<PCollection<KV<K, InputT>>, PCollection<KV<K, Iterable<InputT>>>> transform) {
+  public static <K, InputT>
+      SingleOutputStreamOperator<WindowedValue<KV<K, Iterable<InputT>>>> batchGroupByKey(
+          FlinkStreamingTranslationContext context,
+          PTransform<PCollection<KV<K, InputT>>, PCollection<KV<K, Iterable<InputT>>>> transform) {
 
     Map<Integer, PCollectionView<?>> sideInputTagMapping = new HashMap<>();
     List<PCollectionView<?>> sideInputs = Collections.emptyList();
@@ -372,7 +377,8 @@ public static <K, InputT> SingleOutputStreamOperator<WindowedValue<KV<K, Iterabl
         context.getTypeInfo(context.getOutput(transform));
 
     Coder<Iterable<InputT>> accumulatorCoder = IterableCoder.of(inputKvCoder.getValueCoder());
-    KvCoder<K, Iterable<InputT>> accumKvCoder = KvCoder.of(inputKvCoder.getKeyCoder(), accumulatorCoder);
+    KvCoder<K, Iterable<InputT>> accumKvCoder =
+        KvCoder.of(inputKvCoder.getKeyCoder(), accumulatorCoder);
 
     Coder<WindowedValue<KV<K, Iterable<InputT>>>> windowedAccumCoder =
         WindowedValue.getFullCoder(
@@ -380,50 +386,55 @@ public static <K, InputT> SingleOutputStreamOperator<WindowedValue<KV<K, Iterabl
 
     Coder<WindowedValue<KV<K, Iterable<Iterable<InputT>>>>> outputCoder =
         WindowedValue.getFullCoder(
-            KvCoder.of(inputKvCoder.getKeyCoder(), IterableCoder.of(accumulatorCoder)) , input.getWindowingStrategy().getWindowFn().windowCoder());
+            KvCoder.of(inputKvCoder.getKeyCoder(), IterableCoder.of(accumulatorCoder)),
+            input.getWindowingStrategy().getWindowFn().windowCoder());
 
     TypeInformation<WindowedValue<KV<K, Iterable<Iterable<InputT>>>>> accumulatedTypeInfo =
         new CoderTypeInformation<>(
-          WindowedValue.getFullCoder(
-              KvCoder.of(inputKvCoder.getKeyCoder(), IterableCoder.of(IterableCoder.of(inputKvCoder.getValueCoder()))), input.getWindowingStrategy().getWindowFn().windowCoder()),
+            WindowedValue.getFullCoder(
+                KvCoder.of(
+                    inputKvCoder.getKeyCoder(),
+                    IterableCoder.of(IterableCoder.of(inputKvCoder.getValueCoder()))),
+                input.getWindowingStrategy().getWindowFn().windowCoder()),
             serializablePipelineOptions);
 
     // final aggregation
     WindowDoFnOperator<K, Iterable<InputT>, Iterable<Iterable<InputT>>> finalDoFnOperator =
-          getWindowedAccumulateDoFnOperator(
-              context,
-              transform,
-              accumKvCoder,
-              outputCoder,
-              sideInputTagMapping,
-              sideInputs);
-
-    return
-        getBatchCombinePerKeyOperator(
-          context,
-          input,
-          sideInputTagMapping,
-          sideInputs,
-          windowedAccumCoder,
-          new ConcatenateAsIterable<>(),
-          finalDoFnOperator,
-          accumulatedTypeInfo
-      )
-      .flatMap(new FlattenIterable<>(), outputTypeInfo)
-      .name("concatenate");
+        getWindowedAccumulateDoFnOperator(
+            context, transform, accumKvCoder, outputCoder, sideInputTagMapping, sideInputs);
+
+    return getBatchCombinePerKeyOperator(
+            context,
+            input,
+            sideInputTagMapping,
+            sideInputs,
+            windowedAccumCoder,
+            new ConcatenateAsIterable<>(),
+            finalDoFnOperator,
+            accumulatedTypeInfo)
+        .flatMap(new FlattenIterable<>(), outputTypeInfo)
+        .name("concatenate");
   }
 
-  private static <InputT, K> WindowDoFnOperator<K, Iterable<InputT>, Iterable<Iterable<InputT>>> getWindowedAccumulateDoFnOperator(
-      FlinkStreamingTranslationContext context,
-      PTransform<PCollection<KV<K,InputT>>, PCollection<KV<K, Iterable<InputT>>>> transform,
-      KvCoder<K, Iterable<InputT>> accumKvCoder,
-      Coder<WindowedValue<KV<K, Iterable<Iterable<InputT>>>>> outputCoder,
-      Map<Integer, PCollectionView<?>> sideInputTagMapping,
-      List<PCollectionView<?>> sideInputs) {
+  private static <InputT, K>
+      WindowDoFnOperator<K, Iterable<InputT>, Iterable<Iterable<InputT>>>
+          getWindowedAccumulateDoFnOperator(
+              FlinkStreamingTranslationContext context,
+              PTransform<PCollection<KV<K, InputT>>, PCollection<KV<K, Iterable<InputT>>>>
+                  transform,
+              KvCoder<K, Iterable<InputT>> accumKvCoder,
+              Coder<WindowedValue<KV<K, Iterable<Iterable<InputT>>>>> outputCoder,
+              Map<Integer, PCollectionView<?>> sideInputTagMapping,
+              List<PCollectionView<?>> sideInputs) {
 
-      // Combining fn
-    SystemReduceFn<K, Iterable<InputT>, Iterable<Iterable<InputT>>, Iterable<Iterable<InputT>>, BoundedWindow> reduceFn =
-        SystemReduceFn.buffering(accumKvCoder.getValueCoder());
+    // Combining fn
+    SystemReduceFn<
+            K,
+            Iterable<InputT>,
+            Iterable<Iterable<InputT>>,
+            Iterable<Iterable<InputT>>,
+            BoundedWindow>
+        reduceFn = SystemReduceFn.buffering(accumKvCoder.getValueCoder());
 
     return getWindowedAggregateDoFnOperator(
         context, transform, accumKvCoder, outputCoder, reduceFn, sideInputTagMapping, sideInputs);
@@ -482,8 +493,7 @@ SingleOutputStreamOperator<WindowedValue<KV<K, OutputT>>> batchCombinePerKey(
         windowedAccumCoder,
         combineFn,
         finalDoFnOperator,
-        outputTypeInfo
-    );
+        outputTypeInfo);
   }
 
   @SuppressWarnings({
diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingPipelineTranslator.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingPipelineTranslator.java
index 8f0e6db26dab..0607838987f1 100644
--- a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingPipelineTranslator.java
+++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingPipelineTranslator.java
@@ -27,7 +27,6 @@
 import java.util.Objects;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
-
 import org.apache.beam.runners.flink.adapter.FlinkKey;
 import org.apache.beam.sdk.Pipeline;
 import org.apache.beam.sdk.coders.Coder;
diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingPortablePipelineTranslator.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingPortablePipelineTranslator.java
index 901ab1c672dc..a74be9f7e9e0 100644
--- a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingPortablePipelineTranslator.java
+++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingPortablePipelineTranslator.java
@@ -434,8 +434,7 @@ private <K, V> SingleOutputStreamOperator<WindowedValue<KV<K, Iterable<V>>>> add
         new WorkItemKeySelector<>(inputElementCoder.getKeyCoder());
 
     KeyedStream<WindowedValue<KV<K, V>>, FlinkKey> keyedWorkItemStream =
-        inputDataStream.keyBy(
-            new KvToFlinkKeyKeySelector(inputElementCoder.getKeyCoder()));
+        inputDataStream.keyBy(new KvToFlinkKeyKeySelector(inputElementCoder.getKeyCoder()));
 
     SystemReduceFn<K, V, Iterable<V>, Iterable<V>, BoundedWindow> reduceFn =
         SystemReduceFn.buffering(inputElementCoder.getValueCoder());
@@ -829,8 +828,7 @@ private <InputT, OutputT> void translateExecutableStage(
       }
       if (stateful) {
         keyCoder = ((KvCoder) valueCoder).getKeyCoder();
-        keySelector =
-            new KvToFlinkKeyKeySelector(keyCoder);
+        keySelector = new KvToFlinkKeyKeySelector(keyCoder);
       } else {
         // For an SDF, we know that the input element should be
         // KV<KV<element, KV<restriction, watermarkState>>, size>. We are going to use the element
@@ -844,8 +842,7 @@ private <InputT, OutputT> void translateExecutableStage(
                   valueCoder.getClass().getSimpleName()));
         }
         keyCoder = ((KvCoder) ((KvCoder) valueCoder).getKeyCoder()).getKeyCoder();
-        keySelector =
-            new SdfFlinkKeyKeySelector(keyCoder);
+        keySelector = new SdfFlinkKeyKeySelector(keyCoder);
       }
       inputDataStream = inputDataStream.keyBy(keySelector);
     }
diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingTransformTranslators.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingTransformTranslators.java
index 35a08eb54115..36cf035a33be 100644
--- a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingTransformTranslators.java
+++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingTransformTranslators.java
@@ -105,7 +105,6 @@
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.functions.KeySelector;
 import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.api.java.typeutils.GenericTypeInfo;
 import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
 import org.apache.flink.api.java.typeutils.ValueTypeInfo;
 import org.apache.flink.configuration.Configuration;
@@ -591,8 +590,7 @@ static <InputT, OutputT> void translateParDo(
         // Based on the fact that the signature is stateful, DoFnSignatures ensures
         // that it is also keyed
         keyCoder = ((KvCoder) input.getCoder()).getKeyCoder();
-        keySelector =
-            new KvToFlinkKeyKeySelector<>(keyCoder);
+        keySelector = new KvToFlinkKeyKeySelector<>(keyCoder);
         final PTransform<?, PCollection<InputT>> producer = context.getProducer(input);
         final String previousUrn =
             producer != null
@@ -609,8 +607,7 @@ static <InputT, OutputT> void translateParDo(
       } else if (doFn instanceof SplittableParDoViaKeyedWorkItems.ProcessFn) {
         // we know that it is keyed on byte[]
         keyCoder = ByteArrayCoder.of();
-        keySelector =
-            new WorkItemKeySelector<>(keyCoder);
+        keySelector = new WorkItemKeySelector<>(keyCoder);
         stateful = true;
       }
 
@@ -962,10 +959,7 @@ public void translateNode(
       SingleOutputStreamOperator<WindowedValue<KV<K, Iterable<InputT>>>> outDataStream;
       // Pre-aggregate before shuffle similar to group combine
       if (!context.isStreaming()) {
-        outDataStream =
-            FlinkStreamingAggregationsTranslators.batchGroupByKey(
-                context,
-                transform);
+        outDataStream = FlinkStreamingAggregationsTranslators.batchGroupByKey(context, transform);
       } else {
         // No pre-aggregation in Streaming mode.
         KvToFlinkKeyKeySelector<K, InputT> keySelector =
@@ -1046,8 +1040,7 @@ public void translateNode(
       List<PCollectionView<?>> sideInputs = ((Combine.PerKey) transform).getSideInputs();
 
       KeyedStream<WindowedValue<KV<K, InputT>>, FlinkKey> keyedStream =
-          inputDataStream.keyBy(
-              new KvToFlinkKeyKeySelector<>(keyCoder));
+          inputDataStream.keyBy(new KvToFlinkKeyKeySelector<>(keyCoder));
 
       if (sideInputs.isEmpty()) {
         SingleOutputStreamOperator<WindowedValue<KV<K, OutputT>>> outDataStream;
@@ -1147,8 +1140,7 @@ public void translateNode(
               .name("ToKeyedWorkItem");
 
       KeyedStream<WindowedValue<KeyedWorkItem<K, InputT>>, FlinkKey> keyedWorkItemStream =
-          workItemStream.keyBy(
-              new WorkItemKeySelector<>(inputKvCoder.getKeyCoder()));
+          workItemStream.keyBy(new WorkItemKeySelector<>(inputKvCoder.getKeyCoder()));
 
       context.setOutputDataStream(context.getOutput(transform), keyedWorkItemStream);
     }
diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/adapter/FlinkKey.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/adapter/FlinkKey.java
index cf3871f68a1c..6a5e8d0458f3 100644
--- a/runners/flink/src/main/java/org/apache/beam/runners/flink/adapter/FlinkKey.java
+++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/adapter/FlinkKey.java
@@ -17,6 +17,11 @@
  */
 package org.apache.beam.runners.flink.adapter;
 
+import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import javax.annotation.Nullable;
 import org.apache.beam.runners.flink.translation.types.CoderTypeSerializer;
 import org.apache.beam.runners.flink.translation.wrappers.streaming.FlinkKeyUtils;
 import org.apache.beam.sdk.coders.ByteArrayCoder;
@@ -26,12 +31,6 @@
 import org.apache.flink.core.memory.DataOutputView;
 import org.apache.flink.types.Value;
 
-import javax.annotation.Nullable;
-import java.io.IOException;
-import java.nio.ByteBuffer;
-
-import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
-
 public class FlinkKey implements Value {
 
   private final CoderTypeSerializer<byte[]> serializer;
@@ -77,7 +76,7 @@ public <K> K getKey(Coder<K> coder) {
 
   @Override
   public int hashCode() {
-//    return underlying.hashCode();
+    //    return underlying.hashCode();
     return Hashing.murmur3_128().hashBytes(underlying.array()).asInt();
   }
 
diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/ExecutableStageDoFnOperator.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/ExecutableStageDoFnOperator.java
index c02d7d9c99ea..53e09f3f818c 100644
--- a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/ExecutableStageDoFnOperator.java
+++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/ExecutableStageDoFnOperator.java
@@ -562,8 +562,7 @@ class SdfFlinkTimerInternalsFactory implements TimerInternalsFactory<InputT> {
     @Override
     public TimerInternals timerInternalsForKey(InputT key) {
       try {
-        FlinkKey encodedKey =
-            (FlinkKey) keySelector.getKey(WindowedValue.valueInGlobalWindow(key));
+        FlinkKey encodedKey = (FlinkKey) keySelector.getKey(WindowedValue.valueInGlobalWindow(key));
         return new SdfFlinkTimerInternals(encodedKey);
       } catch (Exception e) {
         throw new RuntimeException("Couldn't get a timer internals", e);
@@ -659,8 +658,7 @@ class SdfFlinkStateInternalsFactory implements StateInternalsFactory<InputT> {
     @Override
     public StateInternals stateInternalsForKey(InputT key) {
       try {
-        FlinkKey encodedKey =
-            (FlinkKey) keySelector.getKey(WindowedValue.valueInGlobalWindow(key));
+        FlinkKey encodedKey = (FlinkKey) keySelector.getKey(WindowedValue.valueInGlobalWindow(key));
         return new SdfFlinkStateInternals(encodedKey);
       } catch (Exception e) {
         throw new RuntimeException("Couldn't get a state internals", e);
@@ -1259,7 +1257,10 @@ void cleanupState(StateInternals stateInternals, Consumer<FlinkKey> keyContextCo
           cleanupTimer.setCleanupTimer(window);
         } else {
           if (LOG.isDebugEnabled()) {
-            LOG.debug("State cleanup for {} {}", Arrays.toString(kv.getKey().getSerializedKey().array()), window);
+            LOG.debug(
+                "State cleanup for {} {}",
+                Arrays.toString(kv.getKey().getSerializedKey().array()),
+                window);
           }
           // No more timers (finally!). Time to clean up.
           for (String userState : userStateNames) {
diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/SdfFlinkKeyKeySelector.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/SdfFlinkKeyKeySelector.java
index 176a585e993d..b316726e74f8 100644
--- a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/SdfFlinkKeyKeySelector.java
+++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/SdfFlinkKeyKeySelector.java
@@ -17,10 +17,7 @@
  */
 package org.apache.beam.runners.flink.translation.wrappers.streaming;
 
-import java.nio.ByteBuffer;
-import org.apache.beam.runners.core.construction.SerializablePipelineOptions;
 import org.apache.beam.runners.flink.adapter.FlinkKey;
-import org.apache.beam.runners.flink.translation.types.CoderTypeInformation;
 import org.apache.beam.sdk.coders.Coder;
 import org.apache.beam.sdk.util.WindowedValue;
 import org.apache.beam.sdk.values.KV;
@@ -32,8 +29,8 @@
 /**
  * {@link KeySelector} that retrieves a key from a {@code KV<KV<element, KV<restriction,
  * watermarkState>>, size>}. This will return the element as encoded by the provided {@link Coder}
- * in a {@link FlinkKey}. This ensures that all key comparisons/hashing happen on the encoded
- * form. Note that the reason we don't use the whole {@code KV<KV<element, KV<restriction,
+ * in a {@link FlinkKey}. This ensures that all key comparisons/hashing happen on the encoded form.
+ * Note that the reason we don't use the whole {@code KV<KV<element, KV<restriction,
  * watermarkState>>, Double>} as the key is when checkpoint happens, we will get different
  * restriction/watermarkState/size, which Flink treats as a new key. Using new key to set state and
  * timer may cause defined behavior.
diff --git a/runners/flink/src/test/java/org/apache/beam/runners/flink/adapter/FlinkKeyTest.java b/runners/flink/src/test/java/org/apache/beam/runners/flink/adapter/FlinkKeyTest.java
index b0fadc0c07eb..649332c1e48f 100644
--- a/runners/flink/src/test/java/org/apache/beam/runners/flink/adapter/FlinkKeyTest.java
+++ b/runners/flink/src/test/java/org/apache/beam/runners/flink/adapter/FlinkKeyTest.java
@@ -17,6 +17,16 @@
  */
 package org.apache.beam.runners.flink.adapter;
 
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThan;
+import static org.hamcrest.Matchers.not;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 import org.apache.beam.sdk.coders.VarIntCoder;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.tuple.Tuple2;
@@ -27,17 +37,6 @@
 import org.hamcrest.core.IsInstanceOf;
 import org.junit.Test;
 
-import java.nio.ByteBuffer;
-import java.nio.charset.StandardCharsets;
-import java.util.Set;
-import java.util.stream.Collectors;
-import java.util.stream.IntStream;
-
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.equalTo;
-import static org.hamcrest.Matchers.greaterThan;
-import static org.hamcrest.Matchers.not;
-
 public class FlinkKeyTest {
   @Test
   public void testIsRecognizedAsValue() {
@@ -48,7 +47,8 @@ public void testIsRecognizedAsValue() {
 
     assertThat(tpe, IsInstanceOf.instanceOf(ValueTypeInfo.class));
 
-    TypeInformation<Tuple2<FlinkKey, byte[]>> tupleTpe = TypeExtractor.getForObject(Tuple2.of(key, bs));
+    TypeInformation<Tuple2<FlinkKey, byte[]>> tupleTpe =
+        TypeExtractor.getForObject(Tuple2.of(key, bs));
     assertThat(tupleTpe, not(IsInstanceOf.instanceOf(GenericTypeInfo.class)));
   }
 
@@ -67,10 +67,11 @@ public void testIsConsistent() {
   private void checkDistribution(int numKeys) {
     int paralellism = 2100;
 
-    Set<Integer> hashcodes = IntStream.range(0, numKeys)
-        .mapToObj(i -> FlinkKey.of(i, VarIntCoder.of()))
-        .map(k -> k.hashCode())
-        .collect(Collectors.toSet());
+    Set<Integer> hashcodes =
+        IntStream.range(0, numKeys)
+            .mapToObj(i -> FlinkKey.of(i, VarIntCoder.of()))
+            .map(k -> k.hashCode())
+            .collect(Collectors.toSet());
 
     Set<Integer> keyGroups =
         hashcodes.stream()
diff --git a/runners/flink/src/test/java/org/apache/beam/runners/flink/streaming/FlinkStateInternalsTest.java b/runners/flink/src/test/java/org/apache/beam/runners/flink/streaming/FlinkStateInternalsTest.java
index 95c255159e6f..2324a262acc0 100644
--- a/runners/flink/src/test/java/org/apache/beam/runners/flink/streaming/FlinkStateInternalsTest.java
+++ b/runners/flink/src/test/java/org/apache/beam/runners/flink/streaming/FlinkStateInternalsTest.java
@@ -40,7 +40,6 @@
 import org.apache.beam.sdk.util.CoderUtils;
 import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.JobID;
-import org.apache.flink.api.java.typeutils.GenericTypeInfo;
 import org.apache.flink.api.java.typeutils.ValueTypeInfo;
 import org.apache.flink.core.fs.CloseableRegistry;
 import org.apache.flink.runtime.jobgraph.JobVertexID;
@@ -212,7 +211,8 @@ public static KeyedStateBackend<FlinkKey> createStateBackend() throws Exception
   private static void changeKey(KeyedStateBackend<FlinkKey> keyedStateBackend)
       throws CoderException {
     keyedStateBackend.setCurrentKey(
-        FlinkKey.of(ByteBuffer.wrap(
-            CoderUtils.encodeToByteArray(StringUtf8Coder.of(), UUID.randomUUID().toString()))));
+        FlinkKey.of(
+            ByteBuffer.wrap(
+                CoderUtils.encodeToByteArray(StringUtf8Coder.of(), UUID.randomUUID().toString()))));
   }
 }
diff --git a/runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperatorTest.java b/runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperatorTest.java
index 7db97769d97a..f0d8816bdeab 100644
--- a/runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperatorTest.java
+++ b/runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperatorTest.java
@@ -376,9 +376,7 @@ public void onProcessingTime(OnTimerContext context) {
 
     OneInputStreamOperatorTestHarness<WindowedValue<Integer>, WindowedValue<String>> testHarness =
         new KeyedOneInputStreamOperatorTestHarness<>(
-            doFnOperator,
-            keySelector,
-            ValueTypeInfo.of(FlinkKey.class));
+            doFnOperator, keySelector, ValueTypeInfo.of(FlinkKey.class));
 
     testHarness.setup(
         new CoderTypeSerializer<>(
@@ -544,9 +542,7 @@ void emitWatermarkIfHoldChanged(long currentWatermarkHold) {
             WindowedValue<KV<String, String>>, WindowedValue<KV<String, String>>>
         testHarness =
             new KeyedOneInputStreamOperatorTestHarness<>(
-                doFnOperator,
-                keySelector,
-                ValueTypeInfo.of(FlinkKey.class));
+                doFnOperator, keySelector, ValueTypeInfo.of(FlinkKey.class));
 
     testHarness.setup();
 
@@ -638,9 +634,7 @@ public void processElement(ProcessContext context) {
 
     OneInputStreamOperatorTestHarness<WindowedValue<Integer>, WindowedValue<String>> testHarness =
         new KeyedOneInputStreamOperatorTestHarness<>(
-            doFnOperator,
-            keySelector,
-            ValueTypeInfo.of(FlinkKey.class));
+            doFnOperator, keySelector, ValueTypeInfo.of(FlinkKey.class));
 
     testHarness.open();
 
@@ -887,9 +881,7 @@ outputTag, coder, new SerializablePipelineOptions(options)),
             Collections.emptyMap());
 
     return new KeyedOneInputStreamOperatorTestHarness<>(
-        doFnOperator,
-        keySelector,
-        ValueTypeInfo.of(FlinkKey.class));
+        doFnOperator, keySelector, ValueTypeInfo.of(FlinkKey.class));
   }
 
   @Test
@@ -943,10 +935,7 @@ outputTag, coder, new SerializablePipelineOptions(FlinkPipelineOptions.defaults(
       // we use a dummy key for the second input since it is considered to be broadcast
       testHarness =
           new KeyedTwoInputStreamOperatorTestHarness<>(
-              doFnOperator,
-              keySelector,
-              null,
-              ValueTypeInfo.of(FlinkKey.class));
+              doFnOperator, keySelector, null, ValueTypeInfo.of(FlinkKey.class));
     }
 
     testHarness.open();
@@ -1035,16 +1024,14 @@ public void processElement(
     TupleTag<KV<String, Long>> outputTag = new TupleTag<>("main-output");
 
     StringUtf8Coder keyCoder = StringUtf8Coder.of();
-    KvToFlinkKeyKeySelector<String, Long> keySelector =
-        new KvToFlinkKeyKeySelector<>(keyCoder);
+    KvToFlinkKeyKeySelector<String, Long> keySelector = new KvToFlinkKeyKeySelector<>(keyCoder);
 
     KvCoder<String, Long> coder = KvCoder.of(keyCoder, VarLongCoder.of());
 
     FullWindowedValueCoder<KV<String, Long>> kvCoder =
         WindowedValue.getFullCoder(coder, windowingStrategy.getWindowFn().windowCoder());
 
-    TypeInformation<FlinkKey> keyCoderInfo =
-        ValueTypeInfo.of(FlinkKey.class);
+    TypeInformation<FlinkKey> keyCoderInfo = ValueTypeInfo.of(FlinkKey.class);
 
     OneInputStreamOperatorTestHarness<
             WindowedValue<KV<String, Long>>, WindowedValue<KV<String, Long>>>
@@ -1666,8 +1653,7 @@ public void finishBundle(FinishBundleContext context) {
   public void testBundleKeyed() throws Exception {
 
     StringUtf8Coder keyCoder = StringUtf8Coder.of();
-    KvToFlinkKeyKeySelector<String, String> keySelector =
-        new KvToFlinkKeyKeySelector<>(keyCoder);
+    KvToFlinkKeyKeySelector<String, String> keySelector = new KvToFlinkKeyKeySelector<>(keyCoder);
     KvCoder<String, String> kvCoder = KvCoder.of(keyCoder, StringUtf8Coder.of());
     WindowedValue.ValueOnlyWindowedValueCoder<KV<String, String>> windowedValueCoder =
         WindowedValue.getValueOnlyCoder(kvCoder);
@@ -2114,8 +2100,7 @@ public void testExactlyOnceBufferingKeyed() throws Exception {
     TupleTag<KV<String, String>> outputTag = new TupleTag<>("main-output");
 
     StringUtf8Coder keyCoder = StringUtf8Coder.of();
-    KvToFlinkKeyKeySelector<String, String> keySelector =
-        new KvToFlinkKeyKeySelector<>(keyCoder);
+    KvToFlinkKeyKeySelector<String, String> keySelector = new KvToFlinkKeyKeySelector<>(keyCoder);
     KvCoder<String, String> kvCoder = KvCoder.of(keyCoder, StringUtf8Coder.of());
     WindowedValue.ValueOnlyWindowedValueCoder<KV<String, String>> windowedValueCoder =
         WindowedValue.getValueOnlyCoder(kvCoder);
@@ -2246,8 +2231,7 @@ public void testFailOnRequiresStableInputAndDisabledCheckpointing() {
     TupleTag<KV<String, String>> outputTag = new TupleTag<>("main-output");
 
     StringUtf8Coder keyCoder = StringUtf8Coder.of();
-    KvToFlinkKeyKeySelector<String, String> keySelector =
-        new KvToFlinkKeyKeySelector<>(keyCoder);
+    KvToFlinkKeyKeySelector<String, String> keySelector = new KvToFlinkKeyKeySelector<>(keyCoder);
     KvCoder<String, String> kvCoder = KvCoder.of(keyCoder, StringUtf8Coder.of());
     WindowedValue.ValueOnlyWindowedValueCoder<KV<String, String>> windowedValueCoder =
         WindowedValue.getValueOnlyCoder(kvCoder);
diff --git a/runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/ExecutableStageDoFnOperatorTest.java b/runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/ExecutableStageDoFnOperatorTest.java
index 68cffda38e36..a0a955aea1d6 100644
--- a/runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/ExecutableStageDoFnOperatorTest.java
+++ b/runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/ExecutableStageDoFnOperatorTest.java
@@ -791,9 +791,7 @@ private void testEnsureDeferredStateCleanupTimerFiring(boolean withCheckpointing
             FlinkKey, WindowedValue<KV<String, Integer>>, WindowedValue<Integer>>
         testHarness =
             new KeyedOneInputStreamOperatorTestHarness(
-                operator,
-                operator.keySelector,
-                ValueTypeInfo.of(FlinkKey.class));
+                operator, operator.keySelector, ValueTypeInfo.of(FlinkKey.class));
 
     testHarness.open();
 
@@ -941,9 +939,7 @@ public void testEnsureStateCleanupOnFinalWatermark() throws Exception {
             FlinkKey, WindowedValue<KV<String, Integer>>, WindowedValue<Integer>>
         testHarness =
             new KeyedOneInputStreamOperatorTestHarness(
-                operator,
-                operator.keySelector,
-                ValueTypeInfo.of(FlinkKey.class));
+                operator, operator.keySelector, ValueTypeInfo.of(FlinkKey.class));
 
     RemoteBundle bundle = Mockito.mock(RemoteBundle.class);
     when(bundle.getInputReceivers())
diff --git a/runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/WindowDoFnOperatorTest.java b/runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/WindowDoFnOperatorTest.java
index 221a8c458886..6380108ddb94 100644
--- a/runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/WindowDoFnOperatorTest.java
+++ b/runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/WindowDoFnOperatorTest.java
@@ -30,7 +30,6 @@
 
 import java.io.ByteArrayOutputStream;
 import java.nio.ByteBuffer;
-
 import org.apache.beam.runners.core.KeyedWorkItem;
 import org.apache.beam.runners.core.SystemReduceFn;
 import org.apache.beam.runners.core.construction.SerializablePipelineOptions;
@@ -75,7 +74,7 @@ public class WindowDoFnOperatorTest {
   public void testRestore() throws Exception {
     // test harness
     KeyedOneInputStreamOperatorTestHarness<
-        FlinkKey, WindowedValue<KV<Long, Long>>, WindowedValue<KV<Long, Long>>>
+            FlinkKey, WindowedValue<KV<Long, Long>>, WindowedValue<KV<Long, Long>>>
         testHarness = createTestHarness(getWindowDoFnOperator(true));
     testHarness.open();