Calculate requested container count based on adding allocated count and outstanding ContainerRequests in Yarn (#3524)

make the requestedContainerCountMap correctly update the container count (#3523)

update the place to decrease the requestedContainerCountMap

Fix running counts for retried flows (#3520)

Allow table to flush after write failure (#3522)

[GOBBLIN-1652]Add more log in the KafkaJobStatusMonitor in case it fails to process one GobblinTrackingEvent (#3513)

* address comments

* use connectionmanager when httpclient is not cloesable

* [GOBBLIN-1652]Add more log in the KafkaJobStatusMonitor in case it fails to process one GobblinTrackingEvent

* add test

* fix test

Co-authored-by: Zihan Li <zihli@zihli-mn2.linkedin.biz>

Make Yarn container and helix instance allocation group by tag (#3519)

[GOBBLIN-1657] Update completion watermark on change_property in IcebergMetadataWriter (#3517)

* [GOBBLIN-1655] Update completion watermark for quiet tables during iceberg registration

* [GOBBLIN-1657] Update completion watermark on change_proerty GMCE

* Added test case to check watermark update on change_property

Co-authored-by: Vikram Bohra <vbohra@vbohra-mn1.linkedin.biz>

[GOBBLIN-1654] Add capacity floor to avoid aggressively requesting resource and small files. (#3515)

* address comments

* use connectionmanager when httpclient is not cloesable

* [GOBBLIN-1654] Add capacity floor to avoid aggressively requesting resource and small files.

* address comments

Co-authored-by: Zihan Li <zihli@zihli-mn2.linkedin.biz>

[GOBBLIN-1653] Shorten job name length if it exceeds 255 characters (#3514)

* Shorten job name length if it exceeds 255 characters (max size for a directory component)

* Address review to account for flow name in hash

[GOBBLIN-1650] Implement flowGroup quotas for the DagManager (#3511)

* Implement flowGroup quotas for the DagManager

* Address review and add comments to tests

* Add guard for double increments on already tracked dags

* Fix tests

[GOBBLIN-1648] Complete use of JDBC `DataSource` 'read-only' validation query by incorporating where previously omitted (#3509)

* Complete use of JDBC `DataSource` 'read-only' validation query by incorporating where previously omitted

* Add logging to demonstrate validation query successfully configured

Add config to set close timeout in HiveRegister (#3512)

add an API in AbstractBaseKafkaConsumerClient to list selected topics (#3501)

[GOBBLIN-1649] Revert gobblin-1633 (#3510)

[GOBBLIN-1639] Prevent metrics reporting if configured, clean up workunit count metric (#3500)

* Move workunit count metrics emitting to Gobblin pipeline, add configuration to prevent metrics reporting if configured

* rename config key

* fix test

* Fix checkstyle and other tests

* Create a custom extensible hook for GaaS metrics on JobLauncher

* Add tests

* Fix failing tests

* Address review

* Address review comment

[GOBBLIN-1647] Add hive commit GTE to HiveMetadataWriter (#3508)

* Add hive commit GTE to HiveMetadataWriter

* Address comment

[GOBBLIN-1633] Fix compaction actions on job failure not retried if compaction succeeds (#3494)

* GOBBLIN-1633

Fix compaction on job failure not retried if compaction succeeds

* Fix typos

[GOBBLIN-1646] Revert yarn container / helix tag group changes (#3507)

Revert "Fix bug when shrinking the container in Yarn service (#3504)"
This reverts commit dd6d910a7e7a90d15258c6c77ebe626ae6d573f9.

Revert "[GOBBLIN-1620]Make yarn container allocation group by helix tag (#3487)"
This reverts commit 3e877951c284ccd68be3634522f9fc2c3d39f81a.

[GOBBLIN-1641] Add meter for sla exceeded flows (#3502)

* Add meter for sla exceeded flows

* fix tests

* Fix test nullpointer

* Address review + augment tests

GOBBLIN-1644 (#3506)

Log assigned participant when helix participant check fails

[GOBBLIN-1645]Change the prefix of dagManager heartbeat to make it consistent with other metrics (#3505)

* address comments

* use connectionmanager when httpclient is not cloesable

* [GOBBLIN-1645]Change the prefix of dagManager heartbeat to make it consistent with other metrics

Co-authored-by: Zihan Li <zihli@zihli-mn2.linkedin.biz>

Fix bug when shrinking the container in Yarn service (#3504)

* Fix bug when shrinking the container in Yarn service

update unit test

update requestedContainerCountMap

* address comment

[GOBBLIN-1637] Add writer, operation, and partition info to failed metadata writer events (#3498)

* Add writer, operation, and partition info to failed metadata writer events

* Add partitionKeys to failure event

* Add all failed writers to event

[GOBBLIN-1638] Fix unbalanced running count metrics due to Azkaban failures (#3499)

* Fix unbalanced running count metrics due to Azkaban failures

* Fix tests

* Address comments

* Rename test

* Update comment with review

[GOBBLIN-1634] Add retries on flow sla kills (#3495)

* Add retries on flow sla kills

* Address review

* Address review comment

[GOBBLIN-1620]Make yarn container allocation group by helix tag (#3487)

* make yarn service aware of helix tag and resource requirment for each workflow so that containers will be assigned to correct task

update test cases

update helix instance tag during task runner initiation

update logs

update test case

* remove lib not used, add test case

address comments

* update test cases

* remove container min and max config

[GOBBLIN-1636] Close DatasetCleaner after clean task (#3497)

* address comments

* use connectionmanager when httpclient is not cloesable

* [GOBBLIN-1636] Close DatasetCleaner after clean task

Co-authored-by: Zihan Li <zihli@zihli-mn2.linkedin.biz>

[GOBBLIN-1635] Avoid loading env configuration when using config store to improve the performance (#3496)

* address comments

* use connectionmanager when httpclient is not cloesable

* [GOBBLIN-1631]Emit heartbeat for dagManagerThread

* [GOBBLIN-1635] Avoid loading env configuration when using config store to improve the performance

* [GOBBLIN-1630] Remove flow level metrics for adhoc flows (#3491)

* Remove emitting metrics for adhoc flows in dagmanager and orchestrator

* Add tests

* Fix tests

* Address comments

* Improve test by validating gauge value

* use data node aliases to figure out data node names before using DMAS (#3493)

* [GOBBLIN-1619] WriterUtils.mkdirsWithRecursivePermission contains race condition and puts unnecessary load on filesystem (#3477)

* [GOBBLIN-1619] Fix race cond. in writerutil mkdirs

* writer util mkdirs previously had race condition when multiple processes
try to create the same parent directory. This causes incorrect
FileNotFoundException
* new implementation does not change the behavior

* Test coverage for retry config

* Wait for file to exist via retry cfg before setting perms

* use user supplied props to create FileSystem in DatasetCleanerTask (#3483)

Co-authored-by: Zihan Li <zihli@zihli-mn2.linkedin.biz>
Co-authored-by: William Lo <lo.william97@gmail.com>
Co-authored-by: Arjun Singh Bora <abora@linkedin.com>
Co-authored-by: Matthew Ho <homatt999@gmail.com>

use user supplied props to create FileSystem in DatasetCleanerTask (#3483)

[GOBBLIN-1619] WriterUtils.mkdirsWithRecursivePermission contains race condition and puts unnecessary load on filesystem (#3477)

* [GOBBLIN-1619] Fix race cond. in writerutil mkdirs

* writer util mkdirs previously had race condition when multiple processes
try to create the same parent directory. This causes incorrect
FileNotFoundException
* new implementation does not change the behavior

* Test coverage for retry config

* Wait for file to exist via retry cfg before setting perms

use data node aliases to figure out data node names before using DMAS (#3493)

[GOBBLIN-1630] Remove flow level metrics for adhoc flows (#3491)

* Remove emitting metrics for adhoc flows in dagmanager and orchestrator

* Add tests

* Fix tests

* Address comments

* Improve test by validating gauge value

[GOBBLIN-1631]Emit heartbeat for dagManagerThread (#3492)

* address comments

* use connectionmanager when httpclient is not cloesable

* [GOBBLIN-1631]Emit heartbeat for dagManagerThread

Co-authored-by: Zihan Li <zihli@zihli-mn2.linkedin.biz>

[GOBBLIN-1624] Refactor quota management, fix various bugs in accounting of running … (#3481)

* Refactor quota management, fix various bugs in accounting of running jobs

* Add javadocs

* Address comments, add metric counts to tests

* Address scenario on startup where quota is decreased

* rename onstartup to onInit

[GOBBLIN-1613] Add metadata writers field to GMCE schema (#3490)

* [GOBBLIN-1613] Add metadata writers field to GMCE schema

* generalize dataset, platform, and table naming
* more test coverage for GMCE writer

* Reverting data.json syntax change.
- Avro doesn't follow regular json syntax

* Clean up random semi colon

* Improve naming

Update README.md

[GOBBLIN-1629] Make GobblinMCEWriter be able to catch error when calculating hive specs (#3489)

* address comments

* use connectionmanager when httpclient is not cloesable

* [GOBBLIN-1629] Make GobblinMCEWriter be able to catch error when calculating hive specs

Co-authored-by: Zihan Li <zihli@zihli-mn2.linkedin.biz>

Add/fix some fields of MetadataWriterFailureEvent (#3485)

[GOBBLIN-1627] provide option to convert datanodes names (#3484)

* provide option to convert datanodes names
address review comments

* address review comments

* address review comments

* address review comment

Add coverage for edge cases when table paths do not exist, check parents (#3482)

[GOBBLIN-1616] Add close connection logic in salseforceSource (#3486)

* [GOBBLIN-1616] Add close connection logic in salseforceSource

* remove unused import

* address comments

* use connectionmanager when httpclient is not cloesable

Co-authored-by: Zihan Li <zihli@zihli-mn2.linkedin.biz>

[GOBBLIN-1621] Make HelixRetriggeringJobCallable emit job skip event when job is dropped due to previous job is running (#3478)

* [GOBBLIN-1621] Make HelixRetriggeringJobCallable emit job skip event when job is dropped due to previous job is running

* address typo

* address comments

* fix checkStyle

* address comments

[GOBBLIN-1623] Fix NPE when try to close RestApiConnector (#3480)

* Fix NPE when try to close RestApiConnector

* fix typo

Clear bad mysql packages from cache in CI/CD machines (#3479)

[GOBBLIN-1617] pass configurations to some HadoopUtils APIs (#3475)

* pass configurations to some HadoopUtils APIs

* address review comments

[GOBBLIN-1616] Make RestApiConnector be able to close the connection finally (#3474)

* [GOBBLIN-1616] Make RestliAPIConnector be able to close the connection finally

* address comments

add config to set log level for any class (#3473)

Fix bug where partitioned tables would always return the wrong equality in paths (#3472)

[GOBBLIN-1602] Change hive table location and partition check to validate using FS r… (#3459)

* Change hive table location and partition check to validate using FS resolvePath to resolve logical paths

* Add tests for Unpartitioned file set

* Address review, add additional throw if locations mismatch for partition location validation

* Fix checkstyles again

* allow partial success policy for workunits

Don't flush on change_property operation (#3467)

Fix case where error GTE is incorrectly sent from MCE writer (#3466)

partial rollback of PR 3464 (#3465)

[GOBBLIN-1604] Throw exception if there are no allocated requests due to lack of res… (#3461)

* Throw exception if there are no allocated requests due to lack of resources

* Fix typo

[GOBBLIN-1603] Throws error if configured when encountering an IO exception while co… (#3460)

* Throws error if configured when encountering an IO exception while collecting copy entities

* Fix checkstyle

[GOBBLIN-1606] change DEFAULT_GOBBLIN_COPY_CHECK_FILESIZE value (#3464)

* change DEFAULT_GOBBLIN_COPY_CHECK_FILESIZE value
do not reuse the metric

* fix unit tests

* address review comments

Upgraded dropwizard metrics library version from 3.2.3 -> 4.1.2 and added a new wrapper class on dropwizard Timer.Context class to handle the code compatibility as the newer version of this class implements AutoClosable instead of Closable. (#3463)

[GOBBLIN-1605] Fix mysql ubuntu download 404 not found for Github Actions CI/CD (#3462)

* fix mysql ubuntu download 404 not found for Github Actions CI/CD

* use fix missing flag instead of updating every package

* use apt-get update

[GOBBLIN-1601] implement ChangePermissionCommitStep (#3457)

* implement ChangePermissionCommitStep
add a configuration to match permission of ancestor directories permissions in source and destination

* address review comments

[GOBBLIN-1598]Fix metrics already exist issue in dag manager (#3454)

* [GOBBLIN-1598]Fix metrics already exist issue in dag manager

* fix typo

* address comments

[GOBBLIN-1597] Add error handling in dagmanager to continue if dag fails to process,… (#3452)

* Add error handling in dagmanager to continue if dag fails to process, make Azkaban client retry on timeouts

* Addressed comments

GOBBLIN-1579 Fail job on hive existing target table location mismatch (#3433)

Co-authored-by: Gnanasekaran <vgnanasekaran@paypal.com>

[GOBBLIN-1596] Ignore already exists exception if the table has already been created… (#3451)

* Ignore already exists exception if the table has already been created by another thread or job entirely

* Address review + add concurrency fix

[GOBBLIn-1595]Fix the dead lock during hive registration (#3450)

Add guard in DagManager for improperly formed SLA (#3449)

[GOBBLIN-1588] Send failure events for write failures when watermark is advanced in MCE writer (#3441)

* Send failure events for write failures when watermark is advanced in MCE writer

* Address comments

[GOBBLIN-1593] Fix bugs in dag manager about metric reporting and job status monitor (#3448)

* [GOBBLIN-1593] Fix bugs in dag manager about metric reporting and job status monitor

* address comments

* fix typo

Fix bug in `JobSpecSerializer` of inadequately preventing access errors (within `MysqlJobCatalog`) (#3447)

[GOBBLIN-1583] Add System level job start SLA (#3437)

* Add System level job start SLA

* Address review and make time unit configurable, improve naming, add comments to tests

* Address review

[GOBBLIN-1592] Make hive copy be able to apply filter on directory (#3446)

[GOBBLIN-1585]GaaS (DagManager) keep retrying a failed job beyond max attempt number (#3439)

* [GOBBLIN-1585]GaaS (DagManager) keep retrying a failed job beyond max attempt number

* address comments

* adding current attempts in job config and cluster events

* add generation into job status

* address comments

* change comments

* address comments

[GOBBLIN-1590] Add low/high watermark information in event emitted by Gobblin cluster (#3443)

[HotFix]Try to fix the mysql dependency issue in Github action (#3445)

* try to fix the dependency issue

* test2

* test3

* test4

* test5

Lazily initialize FileContext and do not store a handle of it so it can be GC'ed when required (#3444)

[GOBBLIN-1584] Add replace record logic for Mysql writer (#3438)

* [GOBBLIN-1584] Add replace record logic for Mysql writer

* Allows user to configure a mysql ingestion job to allow replacement of the record's value for an existing record in the table.
* Also replaces a backward iteration of ResultSet() with forward iteration in MySqlWriterCommands

* remove extra line

* refactor to improve usage of overwrite data field, adjust tests accordingly

* improve documentation and throwing errors

* modify exception for TeradataWriterCommands

Co-authored-by: Urmi Mustafi <umustafi@umustafi-mn1.linkedin.biz>

Bump up code cov version (#3440)

[GOBBLIN-1581] Iterate over Sql ResultSet in Only the Forward Direction (#3435)

* after sql-connector was bumped, exception results from ResultSet.first() being called because it iterates in backward direction of set

Co-authored-by: Urmi Mustafi <umustafi@umustafi-mn1.linkedin.biz>

[GOBBLIN-1575] use reference count in helix manager, so that connect/disconnect are called once and at the right time (#3427)

* make a copy of helix manager, so that each job can keep using it without worrying about disconnect

* address review comments

* address review comments

* use reference count to connect/disconnect only once

* address review comments

[GOBBLIN-1582] Fill low/high watermark info in SourceState for QueryBasedSource (#3436)

* [GOBBLIN-1582] Fill low/high watermark info in SourceState for QueryBasedSource

* add unit test

* address comments to make high/low watermark optional

* Refactor `MysqlSpecStore` into a generalization, `MysqlNonFlowSpecStore` (not limited to `FlowSpec`s), also useable for `TopologySpec`s (#3414)

* Refactor `MysqlSpecStore` into a generalization, `MysqlNonFlowSpecStore` (not limited to `FlowSpec`s), also useable for `TopologySpec`s

* Add missing file, `MysqlNonFlowSpecStoreTest`

* Fixup `MysqlNonFlowSpecStoreTest`

* Simplify implementaiton of `MysqlSpecStore.getSpecsImpl`.

* Rename `MysqlNonFlowSpecStore` to `MysqlBaseFlowSpecStore`.

* Aid maintainers with additional code comments

* [GOBBLIN-1557] Make KafkaSource getFilteredTopics method protected (#3408)

The method was originally private, and it is useful to be able to
override it in subclasses, to redefine how to get topics to be processed.

Change-Id: If94cda2f7a5e65e52e2453427c60f4abb932b3f8

* [GOBBLIN-1567] do not set a custom maxConnLifetime for sql connection (#3418)

* do not set a custom maxConnLifetime for sql connection

* address review comment

* [GOBBLIN-1568] Exponential backoff for Salesforce bulk api polling (#3420)

* Exponential backoff for Salesforce bulk api polling

* Read min and max wait time from prop with default

* set RETENTION_DATASET_ROOT in CleanableIcebergDataset so that any retention job can use this information (#3422)

* [GOBBLIN-1569] Add RDBMS-backed `MysqlJobCatalog`, as alternative to file system storage (#3421)

* Add RDBMS-backed `MysqlJobCatalog`, as alternative to file system storage

* Streamline `JobSpecDeserializer` error handling, on review feedback.

* Refactor `GsonJobSpecSerDe` into a reusable `GenericGsonSpecSerDe`.

* Fix javadoc slipup

* Tag metrics with proxy url if available (#3423)

* remove use of deprecated helix class (#3424)

codestyle changes

* [GOBBLIN-1565]Make GMCEWriter fault tolerant so that one topic failure will not affect other topics in the same container (#3419)

* [GOBBLIN-1565]Make GMCEWriter fault tolerant so that one topic failure will not affect other topics in the same container

* address comments

* change the way we set low watermark to have a better indicate for the watermark range of the snapshot

* address comments

* fix test error

* [GOBBLIN-1552] determine flow status correctly when dag manager is disabled (#3403)

* determine flow status based on the fact if dag manager is enabled
this is needed because when dag manager is not enabled, flow level events are not emitted
and cannot be used to determine flow status. in that case flow status has to be determined
by using job statuses.

store flow status in the FlowStatus

* address review comments

* address review comments

* removed a commented line

* [GOBBLIN-1564] codestyle changes, typo corrections, improved javadoc  and fix a sync… (#3415)

* codestyle changes, typo corrections, improved javadoc  and fix a synchronization issue

* address review comments

* add review comments

* address review comments

* address review comments

* fix bugsFixMain

* do not delete data while dropping a hive table because data is deleted, if needed, separately (#3431)

* [GOBBLIN-1574] Added whitelist for iceberg tables to add new partitio… (#3426)

* [GOBBLIN-1574] Added whitelist for iceberg tables to add new partition column

* fix to failing test case

* Updated IncebergMetadataWriterTest to blacklist the test table from non-completeness tests

* moved dataset name update in tablemetadata

* Added newPartition checks in Table Metadata

* Fixed test case to include new_parition_enabled

Co-authored-by: Vikram Bohra <vbohra@vbohra-mn1.linkedin.biz>
* [GOBBLIN-1577] change the multiplier used in ExponentialWaitStrategy to a reasonable… (#3430)

* change the multiplier used in ExponentialWaitStrategy to 1 second. old multiplier 2ms was retrying too fast for some use cases

* .

* [GOBBLIN-1580] Check table exists instead of call create table directly to make sure table exists (#3432)

* [hotfix] workaround to catch exception when iceberg does not support get metrics for non-union type

* address comments

* [GOBBLIN-1580]Check table exists instead of call create table directly to make sure table exists

* [GOBBLIN-1573]Fix the ClassNotFoundException in streaming test pipeline (#3425)

* [GOBBLIN-1565]Make GMCEWriter fault tolerant so that one topic failure will not affect other topics in the same container

* address comments

* address comments

* [GOBBLIN-1573]Fix the ClassNotFoundException in streaming test pipeline

* [GOBBLIN-1576] skip appending record count to staging file if present… (#3429)

* [GOBBLIN-1576] skip appending record count to staging file if present already

* fixed checkstyle

* fixed method

Co-authored-by: Vikram Bohra <vbohra@vbohra-mn1.linkedin.biz>
* fix the NPE in dagManager

* fix quota check issue in dagManager

* address comments

Co-authored-by: Kip Kohn <ckohn@linkedin.com>
Co-authored-by: Joseph Allemandou <joseph.allemandou@gmail.com>
Co-authored-by: Arjun Singh Bora <abora@linkedin.com>
Co-authored-by: Jiashuo Wang <willwjs@umich.edu>
Co-authored-by: William Lo <lo.william97@gmail.com>
Co-authored-by: vbohra <vbohra@linkedin.com>
Co-authored-by: Vikram Bohra <vbohra@vbohra-mn1.linkedin.biz>

[GOBBLIN-1576] skip appending record count to staging file if present… (#3429)

* [GOBBLIN-1576] skip appending record count to staging file if present already

* fixed checkstyle

* fixed method

Co-authored-by: Vikram Bohra <vbohra@vbohra-mn1.linkedin.biz>

[GOBBLIN-1573]Fix the ClassNotFoundException in streaming test pipeline (#3425)

* [GOBBLIN-1565]Make GMCEWriter fault tolerant so that one topic failure will not affect other topics in the same container

* address comments

* address comments

* [GOBBLIN-1573]Fix the ClassNotFoundException in streaming test pipeline

[GOBBLIN-1580] Check table exists instead of call create table directly to make sure table exists (#3432)

* [hotfix] workaround to catch exception when iceberg does not support get metrics for non-union type

* address comments

* [GOBBLIN-1580]Check table exists instead of call create table directly to make sure table exists

[GOBBLIN-1577] change the multiplier used in ExponentialWaitStrategy to a reasonable… (#3430)

* change the multiplier used in ExponentialWaitStrategy to 1 second. old multiplier 2ms was retrying too fast for some use cases

* .

[GOBBLIN-1574] Added whitelist for iceberg tables to add new partitio… (#3426)

* [GOBBLIN-1574] Added whitelist for iceberg tables to add new partition column

* fix to failing test case

* Updated IncebergMetadataWriterTest to blacklist the test table from non-completeness tests

* moved dataset name update in tablemetadata

* Added newPartition checks in Table Metadata

* Fixed test case to include new_parition_enabled

Co-authored-by: Vikram Bohra <vbohra@vbohra-mn1.linkedin.biz>

do not delete data while dropping a hive table because data is deleted, if needed, separately (#3431)

[GOBBLIN-1564] codestyle changes, typo corrections, improved javadoc  and fix a sync… (#3415)

* codestyle changes, typo corrections, improved javadoc  and fix a synchronization issue

* address review comments

* add review comments

* address review comments

* address review comments

* fix bugsFixMain

[GOBBLIN-1552] determine flow status correctly when dag manager is disabled (#3403)

* determine flow status based on the fact if dag manager is enabled
this is needed because when dag manager is not enabled, flow level events are not emitted
and cannot be used to determine flow status. in that case flow status has to be determined
by using job statuses.

store flow status in the FlowStatus

* address review comments

* address review comments

* removed a commented line

[GOBBLIN-1565]Make GMCEWriter fault tolerant so that one topic failure will not affect other topics in the same container (#3419)

* [GOBBLIN-1565]Make GMCEWriter fault tolerant so that one topic failure will not affect other topics in the same container

* address comments

* change the way we set low watermark to have a better indicate for the watermark range of the snapshot

* address comments

* fix test error

remove use of deprecated helix class (#3424)

codestyle changes

Tag metrics with proxy url if available (#3423)

[GOBBLIN-1569] Add RDBMS-backed `MysqlJobCatalog`, as alternative to file system storage (#3421)

* Add RDBMS-backed `MysqlJobCatalog`, as alternative to file system storage

* Streamline `JobSpecDeserializer` error handling, on review feedback.

* Refactor `GsonJobSpecSerDe` into a reusable `GenericGsonSpecSerDe`.

* Fix javadoc slipup

set RETENTION_DATASET_ROOT in CleanableIcebergDataset so that any retention job can use this information (#3422)

[GOBBLIN-1568] Exponential backoff for Salesforce bulk api polling (#3420)

* Exponential backoff for Salesforce bulk api polling

* Read min and max wait time from prop with default

[GOBBLIN-1567] do not set a custom maxConnLifetime for sql connection (#3418)

* do not set a custom maxConnLifetime for sql connection

* address review comment

[GOBBLIN-1557] Make KafkaSource getFilteredTopics method protected (#3408)

The method was originally private, and it is useful to be able to
override it in subclasses, to redefine how to get topics to be processed.

Change-Id: If94cda2f7a5e65e52e2453427c60f4abb932b3f8

Refactor `MysqlSpecStore` into a generalization, `MysqlNonFlowSpecStore` (not limited to `FlowSpec`s), also useable for `TopologySpec`s (#3414)

* Refactor `MysqlSpecStore` into a generalization, `MysqlNonFlowSpecStore` (not limited to `FlowSpec`s), also useable for `TopologySpec`s

* Add missing file, `MysqlNonFlowSpecStoreTest`

* Fixup `MysqlNonFlowSpecStoreTest`

* Simplify implementaiton of `MysqlSpecStore.getSpecsImpl`.

* Rename `MysqlNonFlowSpecStore` to `MysqlBaseFlowSpecStore`.

* Aid maintainers with additional code comments

[GOBBLIN-1563]Collect more information to analyze the RC for some job cannot emit kafka events to update job status (#3416)

* [GOBBLIN-1563]Collect more information to analyze the RC for some job cannot emit kafka events to update job status

* address comments for GOBBLIN-1561

* address comments

* avoid type cast

[GOBBLIN-1521] Create local mode of streaming kafka job to help user quickly onboard (#3372)

* [GOBBLIN-1521] Create local mode of streaming kafka job to help user quickly onboard

* remove intended change

* update the document

* address comments

* address comments

[GOBBLIN-1559] Support wildcard for input paths (#3410)

* [GOBBLIN-1559] Support wildcard for input paths

* [GOBBLIN-1559] Support wildcard for input paths

* remove new check and allow 'other' to be glob

* go back to adding special case for exact match of this & other

Co-authored-by: Urmi Mustafi <umustafi@umustafi-mn1.linkedin.biz>

[GOBBLIN-1561]Improve error message when flow compilation fails (#3412)

* [GOBBLIN-1561]Improve error message when flow compilation fails

* adddress comments

[GOBBLIN-1556]Add shutdown logic in FsJobConfigurationManager (#3407)

* [hotfix] workaround to catch exception when iceberg does not support get metrics for non-union type

* address comments

* [GOBBLIN-1556]Add shutdown logic in FsJobConfigurationManager