Merge master into time-output-timestamp

.test-infra/jenkins/job_LoadTests_ParDo_Python_37.groovy

@@ -66,6 +66,6 @@ PhraseTriggeringPostCommitBuilder.postCommitJob(
     batchLoadTestJob(delegate, CommonTestProperties.TriggeringContext.PR)
 }
 
-CronJobBuilder.cronJob('beam_LoadTests_Python_ParDo_Dataflow_Batch', 'H 13 * * *', this) {
+CronJobBuilder.cronJob('beam_LoadTests_Python_37_ParDo_Dataflow_Batch', 'H 13 * * *', this) {
     batchLoadTestJob(delegate, CommonTestProperties.TriggeringContext.POST_COMMIT)
 }

buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy

@@ -362,10 +362,10 @@ class BeamModulePlugin implements Plugin<Project> {
     def apex_malhar_version = "3.4.0"
     def aws_java_sdk_version = "1.11.519"
     def aws_java_sdk2_version = "2.5.71"
-    def cassandra_driver_version = "3.6.0"
+    def cassandra_driver_version = "3.8.0"
     def classgraph_version = "4.8.56"
     def generated_grpc_beta_version = "0.44.0"
-    def generated_grpc_ga_version = "1.43.0"
+    def generated_grpc_ga_version = "1.83.0"
     def generated_grpc_dc_beta_version = "0.27.0-alpha"
     def google_auth_version = "0.12.0"
     def google_clients_version = "1.28.0"
@@ -384,7 +384,7 @@ class BeamModulePlugin implements Plugin<Project> {
     def postgres_version = "42.2.2"
     def powermock_version = "2.0.2"
     def proto_google_common_protos_version = "1.17.0"
-    def protobuf_version = "3.6.0"
+    def protobuf_version = "3.11.1"
     def quickcheck_version = "0.8"
     def spark_version = "2.4.4"
     def spark_structured_streaming_version = "2.4.0"
@@ -428,11 +428,10 @@ class BeamModulePlugin implements Plugin<Project> {
         cassandra_driver_core                       : "com.datastax.cassandra:cassandra-driver-core:$cassandra_driver_version",
         cassandra_driver_mapping                    : "com.datastax.cassandra:cassandra-driver-mapping:$cassandra_driver_version",
         classgraph                                  : "io.github.classgraph:classgraph:$classgraph_version",
-        commons_codec                               : "commons-codec:commons-codec:1.10",
+        commons_codec                               : "commons-codec:commons-codec:1.14",
         commons_compress                            : "org.apache.commons:commons-compress:1.19",
         commons_csv                                 : "org.apache.commons:commons-csv:1.4",
-        commons_io_1x                               : "commons-io:commons-io:1.3.2",
-        commons_io_2x                               : "commons-io:commons-io:2.5",
+        commons_io                                  : "commons-io:commons-io:2.6",
         commons_lang3                               : "org.apache.commons:commons-lang3:3.6",
         commons_math3                               : "org.apache.commons:commons-math3:3.6.1",
         error_prone_annotations                     : "com.google.errorprone:error_prone_annotations:2.0.15",
@@ -445,7 +444,7 @@ class BeamModulePlugin implements Plugin<Project> {
         google_api_services_clouddebugger           : "com.google.apis:google-api-services-clouddebugger:v2-rev20181114-$google_clients_version",
         google_api_services_cloudresourcemanager    : "com.google.apis:google-api-services-cloudresourcemanager:v1-rev20181015-$google_clients_version",
         google_api_services_dataflow                : "com.google.apis:google-api-services-dataflow:v1b3-rev20190927-$google_clients_version",
-        google_api_services_pubsub                  : "com.google.apis:google-api-services-pubsub:v1-rev20181213-$google_clients_version",
+        google_api_services_pubsub                  : "com.google.apis:google-api-services-pubsub:v1-rev20191111-$google_clients_version",
         google_api_services_storage                 : "com.google.apis:google-api-services-storage:v1-rev20181109-$google_clients_version",
         google_auth_library_credentials             : "com.google.auth:google-auth-library-credentials:$google_auth_version",
         google_auth_library_oauth2_http             : "com.google.auth:google-auth-library-oauth2-http:$google_auth_version",

model/fn-execution/src/main/proto/beam_fn_api.proto

@@ -426,7 +426,7 @@ message ProcessBundleSplitRequest {
     //
     // Set to 0 to "checkpoint" as soon as possible (keeping as little work as
     // possible and returning the remainder).
-    float fraction_of_remainder = 1;
+    double fraction_of_remainder = 1;
 
     // A set of allowed element indices where the SDK may split. When this is
     // empty, there are no constraints on where to split.
@@ -481,12 +481,12 @@ message ProcessBundleSplitResponse {
     // The last element of the input channel that should be entirely considered
     // part of the primary, identified by its absolute index in the (ordered)
     // channel.
-    int32 last_primary_element = 2;
+    int64 last_primary_element = 2;
 
     // The first element of the input channel that should be entirely considered
     // part of the residual, identified by its absolute index in the (ordered)
     // channel.
-    int32 first_residual_element = 3;
+    int64 first_residual_element = 3;
   }
 
   // Partitions of input data channels into primary and residual elements,

model/pipeline/src/main/proto/beam_runner_api.proto

@@ -679,12 +679,12 @@ message StandardCoders {
 
     // A windowed value coder with parameterized timestamp, windows and pane info.
     // Encodes an element with only the value of the windowed value.
-    // Decodes the value and assign the parameterized timestamp, windows and PaneInfo to the
-    // windowed value
+    // Decodes the value and assigns the parameterized timestamp, windows and pane info to the
+    // windowed value.
     // Components: The element coder and the window coder, in that order
     // The payload of this coder is an encoded windowed value using the
-    // beam:coder:windowed_value:v1 coder parameterized by beam:coder:bytes:v1
-    // elements coder and the window coder that this param_windowed_value uses.
+    // beam:coder:windowed_value:v1 coder parameterized by a beam:coder:bytes:v1
+    // element coder and the window coder that this param_windowed_value coder uses.
     PARAM_WINDOWED_VALUE = 14 [(beam_urn) = "beam:coder:param_windowed_value:v1"];
 
     // Encodes an iterable of elements, some of which may be stored elsewhere.
@@ -1089,8 +1089,8 @@ message WireCoderSetting {
   // the URN is beam:coder:windowed_value:v1, this may be omitted. If the URN is
   // beam:coder:param_windowed_value:v1, the payload is an encoded windowed
   // value using the beam:coder:windowed_value:v1 coder parameterized by
-  // beam:coder:bytes:v1 elements coder and the window coder that this
-  // param_windowed_value uses.
+  // a beam:coder:bytes:v1 element coder and the window coder that this
+  // param_windowed_value coder uses.
   bytes payload = 2;
 }
 

release/src/main/scripts/publish_docker_images.sh

@@ -24,43 +24,78 @@
 
 set -e
 
-source release/src/main/scripts/build_release_candidate.sh
+PYTHON_VER=("python2.7" "python3.5" "python3.6" "python3.7")
+FLINK_VER=("$(ls -1 runners/flink | awk '/^[0-9]+\.[0-9]+$/{print}')")
 
 echo "Publish SDK docker images to Docker Hub."
+
+echo "================Setting Up Environment Variables==========="
+echo "Which release version are you working on: "
+read RELEASE
+
+echo "================Setting Up RC candidate Variables==========="
+echo "From which RC candidate do you create publish docker image? (ex: rc0, rc1) "
+read RC_VERSION
+
+echo "================Confirmimg Release and RC version==========="
+echo "We are using ${RC_VERSION} to create docker images for ${RELEASE}."
 echo "Do you want to proceed? [y|N]"
 read confirmation
 if [[ $confirmation = "y" ]]; then
-  echo "============Publishing SDK docker images on docker hub========="
-  cd ~
-  if [[ -d ${LOCAL_PYTHON_STAGING_DIR} ]]; then
-    rm -rf ${LOCAL_PYTHON_STAGING_DIR}
-  fi
-  mkdir -p ${LOCAL_PYTHON_STAGING_DIR}
-  cd ${LOCAL_PYTHON_STAGING_DIR}
-
-  echo '-------------------Cloning Beam Release Branch-----------------'
-  git clone ${GIT_REPO_URL}
-  cd ${BEAM_ROOT_DIR}
-  git checkout ${RELEASE_BRANCH}
 
   echo '-------------------Generating and Pushing Python images-----------------'
-  ./gradlew :sdks:python:container:buildAll -Pdocker-tag=${RELEASE}
   for ver in "${PYTHON_VER[@]}"; do
+     # Pull varified RC from dockerhub.
+     docker pull apachebeam/${ver}_sdk:${RELEASE}_${RC_VERSION}
+
+     # Tag with ${RELEASE} and push to dockerhub.
+     docker tag apachebeam/${ver}_sdk:${RELEASE}_${RC_VERSION} apachebeam/${ver}_sdk:${RELEASE}
      docker push apachebeam/${ver}_sdk:${RELEASE}
-     docker tag apachebeam/${ver}_sdk:${RELEASE} apachebeam/${ver}_sdk:latest
+
+     # Tag with latest and push to dockerhub.
+     docker tag apachebeam/${ver}_sdk:${RELEASE}_${RC_VERSION} apachebeam/${ver}_sdk:latest
      docker push apachebeam/${ver}_sdk:latest
+
+      # Cleanup images from local
+     docker rmi -f apachebeam/${ver}_sdk:${RELEASE}_${RC_VERSION}
+     docker rmi -f apachebeam/${ver}_sdk:${RELEASE}
+     docker rmi -f apachebeam/${ver}_sdk:latest
   done
 
   echo '-------------------Generating and Pushing Java images-----------------'
-  ./gradlew :sdks:java:container:dockerPush -Pdocker-tag=${RELEASE}
-  docker tag apachebeam/java_sdk:${RELEASE} apachebeam/java_sdk:latest
+  # Pull varified RC from dockerhub.
+  docker pull apachebeam/java_sdk:${RELEASE}_${RC_VERSION}
+
+  # Tag with ${RELEASE} and push to dockerhub.
+  docker tag apachebeam/java_sdk:${RELEASE}_${RC_VERSION} apachebeam/java_sdk:${RELEASE}
+  docker push apachebeam/java_sdk:${RELEASE}
+
+  # Tag with latest and push to dockerhub.
+  docker tag apachebeam/java_sdk:${RELEASE}_${RC_VERSION} apachebeam/java_sdk:latest
   docker push apachebeam/java_sdk:latest
 
+  # Cleanup images from local
+  docker rmi -f apachebeam/java_sdk:${RELEASE}_${RC_VERSION}
+  docker rmi -f apachebeam/java_sdk:${RELEASE}
+  docker rmi -f apachebeam/java_sdk:latest
+
   echo '-------------------Generating and Pushing Go images-----------------'
-  ./gradlew :sdks:go:container:dockerPush -Pdocker-tag=${RELEASE}
-  docker tag apachebeam/go_sdk:${RELEASE} apachebeam/go_sdk:latest
+  # Pull varified RC from dockerhub.
+  docker pull apachebeam/go_sdk:${RELEASE}_${RC_VERSION}
+
+  # Tag with ${RELEASE} and push to dockerhub.
+  docker tag apachebeam/go_sdk:${RELEASE}_${RC_VERSION} apachebeam/go_sdk:${RELEASE}
+  docker push apachebeam/go_sdk:${RELEASE}
+
+  # Tag with latest and push to dockerhub.
+  docker tag apachebeam/go_sdk:${RELEASE}_${RC_VERSION} apachebeam/go_sdk:latest
   docker push apachebeam/go_sdk:latest
 
+  # Cleanup images from local
+  docker rmi -f apachebeam/go_sdk:${RELEASE}_${RC_VERSION}
+  docker rmi -f apachebeam/go_sdk:${RELEASE}
+  docker rmi -f apachebeam/go_sdk:latest
+
   echo '-------------Generating and Pushing Flink job server images-------------'
   echo "Building containers for the following Flink versions:" "${FLINK_VER[@]}"
   for ver in "${FLINK_VER[@]}"; do
@@ -71,20 +106,6 @@ if [[ $confirmation = "y" ]]; then
      docker push "${FLINK_IMAGE_NAME}:latest"
   done
 
-  rm -rf ~/${PYTHON_ARTIFACTS_DIR}
-
-  echo "-------------------Clean up SDK docker images at local-------------------"
-  for ver in "${PYTHON_VER[@]}"; do
-    docker rmi -f apachebeam/${ver}_sdk:${RELEASE}
-    docker rmi -f apachebeam/${ver}_sdk:latest
-  done
-
-  docker rmi -f apachebeam/java_sdk:${RELEASE}
-  docker rmi -f apachebeam/java_sdk:latest
-
-  docker rmi -f apachebeam/go_sdk:${RELEASE}
-  docker rmi -f apachebeam/go_sdk:latest
-
   for ver in "${FLINK_VER[@]}"; do
     FLINK_IMAGE_NAME=apachebeam/flink${ver}_job_server
     docker rmi -f "${FLINK_IMAGE_NAME}:${RELEASE}"

runners/apex/src/main/java/org/apache/beam/runners/apex/translation/operators/ApexParDoOperator.java

@@ -385,6 +385,7 @@ public void fireTimer(Object key, Collection<TimerData> timerDataSet) {
       BoundedWindow window = ((WindowNamespace<?>) namespace).getWindow();
       pushbackDoFnRunner.onTimer(
           timerData.getTimerId(),
+          timerData.getTimerFamilyId(),
           window,
           timerData.getTimestamp(),
           timerData.getOutputTimestamp(),